From 5cebc1f9293263c459e4c563c1bb501992d895be Mon Sep 17 00:00:00 2001 From: zk Date: Fri, 3 Jul 2026 17:27:18 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A1=A5=E5=85=85=E5=85=AC=E5=8F=B8log?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env | 7 ++++++ .env.prod | 7 ++++++ .env.test | 7 ++++++ app/config/settings.py | 10 ++++++++ app/core/oss.py | 42 +++++++++++++++++++++++++++++++ app/models/pg/app_url_list.py | 35 ++++++++++++++++++++++++++ app/services/job_clean_service.py | 37 ++++++++++++++++++++++++--- requirements.txt | 3 ++- 8 files changed, 143 insertions(+), 5 deletions(-) create mode 100644 app/core/oss.py create mode 100644 app/models/pg/app_url_list.py diff --git a/.env b/.env index 913a6a5..c2baac1 100644 --- a/.env +++ b/.env @@ -39,3 +39,10 @@ JOB_EXPIRE_DAYS=7 # 日志 LOGGING_LEVEL=INFO + +# 阿里云 OSS +OSS_ACCESS_KEY_ID=LTAI5tEdLKKQUKhTyUpfH5Mk +OSS_ACCESS_KEY_SECRET=RjUdTrq0V5qA4b3BUElNhXqs3ZLp5k +OSS_ENDPOINT=oss-cn-guangzhou.aliyuncs.com +OSS_BUCKET=offerpie +OSS_DOMAIN=https://offerpie.oss-cn-guangzhou.aliyuncs.com diff --git a/.env.prod b/.env.prod index d5caed4..b82ac67 100644 --- a/.env.prod +++ b/.env.prod @@ -33,3 +33,10 @@ JOB_EXPIRE_DAYS=7 # 日志 LOGGING_LEVEL=INFO + +# 阿里云 OSS +OSS_ACCESS_KEY_ID=LTAI5tEdLKKQUKhTyUpfH5Mk +OSS_ACCESS_KEY_SECRET=RjUdTrq0V5qA4b3BUElNhXqs3ZLp5k +OSS_ENDPOINT=oss-cn-guangzhou.aliyuncs.com +OSS_BUCKET=offerpie +OSS_DOMAIN=https://offerpie.oss-cn-guangzhou.aliyuncs.com diff --git a/.env.test b/.env.test index badb7b6..bf748c6 100644 --- a/.env.test +++ b/.env.test @@ -34,3 +34,10 @@ JOB_EXPIRE_DAYS=7 # 日志 LOGGING_LEVEL=INFO + +# 阿里云 OSS +OSS_ACCESS_KEY_ID=LTAI5tEdLKKQUKhTyUpfH5Mk +OSS_ACCESS_KEY_SECRET=RjUdTrq0V5qA4b3BUElNhXqs3ZLp5k +OSS_ENDPOINT=oss-cn-guangzhou.aliyuncs.com +OSS_BUCKET=offerpie +OSS_DOMAIN=https://offerpie.oss-cn-guangzhou.aliyuncs.com diff --git a/app/config/settings.py b/app/config/settings.py index e95b101..719d3f5 100644 --- a/app/config/settings.py +++ b/app/config/settings.py @@ -48,6 +48,16 @@ class Settings(BaseSettings): # ──────────── 岗位下架参数 ──────────── job_expire_days: int = 7 + # ──────────── 阿里云 OSS ──────────── + oss_access_key_id: str = "" + oss_access_key_secret: str = "" + oss_endpoint: str = "oss-cn-guangzhou.aliyuncs.com" + oss_bucket: str = "offerpie" + # 固定上传目录（对象 key 前缀） + oss_upload_dir: str = "company/logo" + # 下载访问域名 + oss_domain: str = "https://offerpie.oss-cn-guangzhou.aliyuncs.com" + # ──────────── 日志 ──────────── logging_level: str = "INFO" log_file_name: str = "cleaner.log" diff --git a/app/core/oss.py b/app/core/oss.py new file mode 100644 index 0000000..dedcddd --- /dev/null +++ b/app/core/oss.py @@ -0,0 +1,42 @@ +"""阿里云 OSS 上传：往固定目录上传，返回下载地址""" + +import asyncio +import base64 +import os +from uuid import uuid4 + +import oss2 + +from app.config import settings +from app.core.logger import log + +_auth = oss2.Auth(settings.oss_access_key_id, settings.oss_access_key_secret) +_bucket = oss2.Bucket(_auth, f"https://{settings.oss_endpoint}", settings.oss_bucket) + + +async def upload(data: bytes, file_name: str) -> str: + """上传字节内容到固定目录，返回下载地址""" + key = f"{settings.oss_upload_dir}/{uuid4().hex[:18]}{os.path.splitext(file_name)[1]}" + await asyncio.to_thread(_bucket.put_object, key, data) + url = f"{settings.oss_domain}/{key}" + log.info("OSS 上传成功 [{}] -> {}", file_name, url) + return url + + +async def upload_base64(b64: str, file_name: str = "image.png") -> str | None: + """上传 base64 图片到固定目录，返回下载地址；空值或解码失败返回 None + + :param b64: base64 字符串，兼容 data URI 前缀（data:image/png;base64,xxx） + :param file_name: 用于取后缀，默认 png + """ + if not b64 or not b64.strip(): + return None + # 去掉 data URI 前缀 + if "," in b64 and b64.strip().startswith("data:"): + b64 = b64.split(",", 1)[1] + try: + data = base64.b64decode(b64) + except Exception as e: + log.warning("base64 解码失败: {}", e) + return None + return await upload(data, file_name) diff --git a/app/models/pg/app_url_list.py b/app/models/pg/app_url_list.py new file mode 100644 index 0000000..d49f847 --- /dev/null +++ b/app/models/pg/app_url_list.py @@ -0,0 +1,35 @@ +"""PostgreSQL: app_url_list 表模型""" + +from datetime import datetime +from typing import Optional + +from sqlalchemy import BigInteger, DateTime, Integer, String, Text +from sqlalchemy.orm import Mapped, mapped_column + +from app.core.database import PgBase + + +class AppUrlList(PgBase): + """爬虫 URL 任务列表""" + + __tablename__ = "app_url_list" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True, comment="自增主键") + status: Mapped[str] = mapped_column(String(32), default="pending", nullable=False, comment="任务状态") + input_company_name: Mapped[str] = mapped_column(String(255), nullable=False, comment="输入公司名") + input_url: Mapped[Optional[str]] = mapped_column(String(1024), comment="输入URL") + output_url: Mapped[Optional[str]] = mapped_column(String(1024), comment="输出URL") + post_login: Mapped[Optional[int]] = mapped_column(Integer, comment="发布登录") + crawl_login: Mapped[Optional[int]] = mapped_column(Integer, comment="爬取登录") + type_sz: Mapped[Optional[int]] = mapped_column(Integer, comment="类型sz") + type_sx: Mapped[Optional[int]] = mapped_column(Integer, comment="类型sx") + crawlok_at: Mapped[Optional[str]] = mapped_column(String(32), comment="-2没有数据或者数据全部过期或者数据缺失 -3 过滤后没数据， -4无法修复，") + pyname: Mapped[Optional[str]] = mapped_column(String(255), comment="脚本名字") + type_xz: Mapped[Optional[int]] = mapped_column(Integer, comment="类型xz") + error_message: Mapped[Optional[str]] = mapped_column(Text, comment="错误信息") + logo: Mapped[Optional[str]] = mapped_column(Text, comment="公司logo (base64编码的64x64 PNG图片)") + started_at: Mapped[Optional[datetime]] = mapped_column(DateTime, comment="开始时间") + finished_at: Mapped[Optional[datetime]] = mapped_column(DateTime, comment="完成时间") + created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, comment="创建时间") + updated_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, comment="更新时间") + output_company_name: Mapped[Optional[str]] = mapped_column(String(255), comment="输出公司名") diff --git a/app/services/job_clean_service.py b/app/services/job_clean_service.py index f7fb0a4..07d870e 100644 --- a/app/services/job_clean_service.py +++ b/app/services/job_clean_service.py @@ -10,6 +10,7 @@ from sqlalchemy import text, insert from app.config import settings from app.core.database import PgSession, MysqlSession from app.core.logger import log +from app.core.oss import upload_base64 from app.ai.model_config import JobCleanModel from app.ai.prompts import JOB_STRUCTURE_SYSTEM, MAJOR_MATCH_SYSTEM, SKILL_EXTRACT_SYSTEM from app.models.mysql.job import Job @@ -134,7 +135,7 @@ async def _do_clean(data: dict) -> None: # 公司处理 company_short_name = result.get("companyShortName") or data.get("company") or "" - company_id = await _find_or_create_company(company_short_name) + company_id = await _find_or_create_company(company_short_name, data.get("urllistid")) # 地区处理 region_codes = [] @@ -284,8 +285,8 @@ async def _find_or_create_skill_tag(name: str) -> int | None: return row.scalar() -async def _find_or_create_company(short_name: str) -> int: - """查找或创建公司（加锁防并发重复）""" +async def _find_or_create_company(short_name: str, urllistid: int | None = None) -> int: + """查找或创建公司（加锁防并发重复）；新建公司时若有 logo 则上传 OSS 并回填地址""" async with _company_lock: async with MysqlSession() as mysql: row = await mysql.execute( @@ -309,7 +310,35 @@ async def _find_or_create_company(short_name: str) -> int: ) ) await mysql.commit() - return company_id + + # 锁外处理 logo：仅新建公司时执行，上传是网络IO，不阻塞其他协程；失败不影响主流程 + try: + logo_b64 = await _get_logo_base64(urllistid) + if logo_b64: + logo_url = await upload_base64(logo_b64, "logo.png") + if logo_url: + async with MysqlSession() as mysql: + await mysql.execute( + text("UPDATE bg_company SET logo_url = :url, update_time = :t WHERE id = :id"), + {"url": logo_url, "t": datetime.now(), "id": company_id}, + ) + await mysql.commit() + except Exception as e: + log.warning("[company={}] logo 上传失败: {}", company_id, e) + + return company_id + + +async def _get_logo_base64(urllistid: int | None) -> str | None: + """从 PG app_url_list 按 urllistid 读取 logo（base64）""" + if not urllistid: + return None + async with PgSession() as pg: + row = await pg.execute( + text("SELECT logo FROM app_url_list WHERE id = :id"), + {"id": urllistid}, + ) + return row.scalar() async def _update_pg_status(data_id: int, status: str) -> None: diff --git a/requirements.txt b/requirements.txt index 58e4838..ad819da 100644 --- a/requirements.txt +++ b/requirements.txt @@ -13,4 +13,5 @@ langchain-core>=0.3 # 工具 loguru>=0.7 snowflake-id>=1.0 -json-repair>=0.30 \ No newline at end of file +json-repair>=0.30 +oss2>=2.18 \ No newline at end of file