zk
112 lines
5.1 KiB
Python
112 lines
5.1 KiB
Python
"""浏览器插件 Service
|
||
|
||
表单填写AI生成：查简历 + 岗位 + agent_config → 调 AI 生成答案。
|
||
"""
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.ai.browser_plug.form_filler import generate_form_answer
|
||
from app.models.job import Job
|
||
from app.models.job_agent_config import JobAgentConfig
|
||
from app.services import customize_resume_store
|
||
|
||
|
||
class BrowserPlugService:
|
||
|
||
def __init__(self, session: AsyncSession):
|
||
self.session = session
|
||
|
||
async def form_fill_answer(self, user_id: int, job_id: int, label: str, reference: str | None, field_type: str) -> str:
|
||
"""生成表单字段的填写内容"""
|
||
# 1. 获取简历数据（有定制简历用定制，没有自动走默认）
|
||
resume_data = await customize_resume_store.get(user_id, job_id)
|
||
resume_text = self._serialize_resume(resume_data)
|
||
|
||
# 2. 获取岗位信息
|
||
job_text = await self._get_job_text(job_id)
|
||
|
||
# 3. 获取 agent_config
|
||
agent_config_text = await self._get_agent_config_text(user_id)
|
||
|
||
# 4. 调 AI 生成
|
||
return await generate_form_answer(resume_text, job_text, agent_config_text, label, reference, field_type)
|
||
|
||
async def _get_job_text(self, job_id: int) -> str:
|
||
"""获取岗位信息文本"""
|
||
if job_id == 0:
|
||
return "未指定岗位"
|
||
result = await self.session.execute(select(Job).where(Job.id == job_id))
|
||
job = result.scalar_one_or_none()
|
||
if not job:
|
||
return "未指定岗位"
|
||
parts = []
|
||
if job.title:
|
||
parts.append(f"岗位名称：{job.title}")
|
||
if job.description:
|
||
parts.append(f"岗位描述：{job.description}")
|
||
if job.requirement:
|
||
parts.append(f"岗位要求：{job.requirement}")
|
||
return "\n".join(parts) if parts else "未指定岗位"
|
||
|
||
async def _get_agent_config_text(self, user_id: int) -> str:
|
||
"""获取 agent_config 预设答案文本"""
|
||
result = await self.session.execute(
|
||
select(JobAgentConfig).where(JobAgentConfig.user_id == user_id))
|
||
config = result.scalar_one_or_none()
|
||
if not config:
|
||
return "无预设配置"
|
||
parts = []
|
||
if config.accept_dept_transfer:
|
||
parts.append(f"是否接受部门调剂：{config.accept_dept_transfer}")
|
||
if config.accept_location_transfer:
|
||
parts.append(f"是否接受地点调剂：{config.accept_location_transfer}")
|
||
if config.interview_type:
|
||
parts.append(f"可参加面试方式：{'、'.join(config.interview_type)}")
|
||
if config.languages:
|
||
lang_strs = [f"{l.get('language', '')}({l.get('proficiency', '')})" for l in config.languages]
|
||
parts.append(f"语言能力：{'、'.join(lang_strs)}")
|
||
if config.available_date:
|
||
parts.append(f"预计到岗时间：{config.available_date}")
|
||
if config.intern_days_per_week:
|
||
parts.append(f"每周实习天数：{config.intern_days_per_week}")
|
||
if config.intern_duration:
|
||
parts.append(f"预计实习时长：{config.intern_duration}")
|
||
return "\n".join(parts) if parts else "无预设配置"
|
||
|
||
@staticmethod
|
||
def _serialize_resume(data: dict) -> str:
|
||
"""将简历 dict 序列化为文本"""
|
||
if not data:
|
||
return "暂无简历信息"
|
||
parts = []
|
||
resume = data.get("resume", {})
|
||
if resume.get("name"):
|
||
parts.append(f"姓名：{resume['name']}")
|
||
if resume.get("email"):
|
||
parts.append(f"邮箱：{resume['email']}")
|
||
if resume.get("mobileNumber"):
|
||
parts.append(f"手机：{resume['mobileNumber']}")
|
||
if resume.get("city"):
|
||
parts.append(f"城市：{resume['city']}")
|
||
if resume.get("skills"):
|
||
parts.append(f"技能：{'、'.join(resume['skills'])}")
|
||
if resume.get("certificates"):
|
||
parts.append(f"证书：{'、'.join(resume['certificates'])}")
|
||
if resume.get("summary"):
|
||
parts.append(f"个人概述：{resume['summary']}")
|
||
for section, title in [("education", "教育经历"), ("work", "工作经历"), ("internship", "实习经历"), ("project", "项目经历"), ("competition", "竞赛经历")]:
|
||
items = data.get(section, [])
|
||
if items:
|
||
parts.append(f"{title}：")
|
||
for item in items:
|
||
if section == "education":
|
||
parts.append(f" - {item.get('school', '')} {item.get('major', '')} {item.get('degree', '')}")
|
||
elif section in ("work", "internship"):
|
||
parts.append(f" - {item.get('companyName', '')} {item.get('position', '')}")
|
||
elif section == "project":
|
||
parts.append(f" - {item.get('projectName', '')} {item.get('role', '')}")
|
||
elif section == "competition":
|
||
parts.append(f" - {item.get('competitionName', '')} {item.get('award', '')}")
|
||
return "\n".join(parts) if parts else "暂无简历信息"
|