📅 2026-07-27
🏷️
Playwright
IIFE
自动化
Playwright + IIFE 实现 DevOps 自动化操作
背景
在技术经理工作台中,需要实现 DevOps 平台的自动化操作,包括自动登录、拉取日报、创建任务等。由于 DevOps 平台是一个 SPA(单页应用),传统的 HTTP 请求无法直接触发前端 JavaScript 逻辑,因此需要一种能够在浏览器环境中执行的方案。
技术选型
经过调研,选择了 Playwright + IIFE(Immediately Invoked Function Expression) 的方案:
- Playwright:微软出品的浏览器自动化框架,支持 Chromium、Firefox、WebKit,API 简洁,适合在 Python 后端中嵌入使用。
- IIFE:将需要执行的 JavaScript 代码封装为立即执行函数,通过 Playwright 的
page.evaluate() 方法在前端执行。
实现方案
整体架构分为三层:
┌─────────────────┐
│ Python 后端 │ ← FastAPI 路由
│ (main.py) │
├─────────────────┤
│ Node.js 执行器 │ ← 独立进程,通过 HTTP 通信
│ (iife_server.js)│
├─────────────────┤
│ Playwright 引擎 │ ← 浏览器自动化
│ (Chromium) │
└─────────────────┘
1. 持久化浏览器实例
为了避免每次登录都启动 Chromium,我们在 auth.py 中维护了一个持久化的浏览器实例:
# 持久化 Playwright 浏览器(线程安全)
_pw_browser = None
_pw_lock = threading.Lock()
def _get_browser():
global _pw_browser
if _pw_browser is not None:
try:
_pw_browser.contexts # 检测浏览器是否活着
return _pw_browser
except Exception:
_pw_browser = None
# 启动新浏览器...
2. Cookie 持久化
登录成功后,将 DevOps Cookie 持久化到本地 JSON 文件中,避免重复登录。每次启动时自动读取 Cookie 恢复会话:
state = {"cookies": devops_cookies, "username": username, "login_time": time.time()}
COOKIE_FILE.write_text(json.dumps(state), encoding="utf-8")
3. IIFE 执行
对于需要在 DevOps 页面内执行的复杂操作(如批量操作、数据提取),通过独立的 Node.js 进程执行 IIFE:
// iife_server.js
const { chromium } = require('playwright');
const browser = await chromium.launch({ headless: true });
const page = await browser.newPage();
await page.goto('https://devops.ctjsoft.com');
const result = await page.evaluate(`(${iifeCode})()`);
经验总结
- 线程安全:Playwright 实例需要加锁保护,避免多请求同时操作导致崩溃。
- 超时处理:登录操作设置 60 秒超时,避免长时间阻塞。
- 降级策略:Playwright 登录失败时,降级使用本地 Cookie 或手动输入。
- 会话检测:每次请求前检查 Cookie 有效性,过期自动提示重新登录。
Playwright 的同步 API(sync_api)在 Python 中与 FastAPI 的异步框架结合时,需要通过 threading 在独立线程中执行,避免 greenlet 冲突。
📅 2026-07-27
FastAPI
Python
架构
FastAPI 在团队管理平台中的实践
为什么选择 FastAPI
FastAPI 作为高性能 Python Web 框架,具备以下优势:
- 异步原生:基于 Starlette,天然支持 async/await,适合 IO 密集型操作(数据库查询、HTTP 请求)。
- 自动文档:自动生成 OpenAPI 文档(Swagger UI + ReDoc),方便调试和团队协作。
- 类型安全:基于 Pydantic 的参数校验,减少运行时错误。
- 中间件机制:灵活的自定义中间件,便于实现 TraceId、权限、日志等横切关注点。
项目架构
workbench/
├── backend/
│ ├── main.py # FastAPI 应用入口,注册路由和中间件
│ ├── config.py # 数据库配置、连接池
│ ├── middleware.py # TraceId、权限中间件
│ ├── routers/ # API 路由模块
│ │ ├── auth.py # 登录鉴权
│ │ ├── daily_report.py # 日报相关
│ │ ├── work_items.py # DevOps 事项
│ │ ├── tasks.py # 任务管理
│ │ └── ...
│ ├── services/ # 业务逻辑层
│ │ ├── automation.py # 定时任务调度
│ │ ├── config_service.py# 配置服务
│ │ ├── devops_api.py # DevOps API 封装
│ │ └── llm_analysis.py # AI 分析服务
│ └── schema.sql # 数据库建表语句
├── frontend/
│ ├── index.html # 主页面
│ ├── blog.html # 技术博客
│ ├── css/style.css # 样式
│ └── js/app.js # 前端应用逻辑
└── README.md
数据库连接池设计
使用自定义连接池替代 SQLAlchemy,轻量且高效:
# 线程安全连接池
_POOL_MIN = 3
_POOL_MAX = 50
_pool_lock = threading.Lock()
_connection_pool = Queue(maxsize=_POOL_MAX)
def get_db():
"""从连接池获取连接,用完自动归还"""
try:
conn = _connection_pool.get_nowait()
conn.ping(reconnect=True)
return conn
except Empty:
return _create_connection()
def return_db(conn):
"""归还连接到池中"""
if conn is None: return
_connection_pool.put_nowait(conn)
中间件链
中间件执行顺序(从外到内):
- TraceIdMiddleware — 生成唯一追踪 ID,记录请求日志
- PermissionMiddleware — 注入角色权限
- set_user_context — 设置请求上下文
- no_cache_html — 禁用 HTML 页面缓存
注意事项
- 中间件顺序:
add_middleware 后添加的为外层,先执行。TraceId 需要放在最外层以保证最早捕获到请求。
- 异常处理:全局异常处理器捕获所有未处理的异常,返回统一的
{ok, error, trace_id} 格式,不暴露堆栈信息。
- 热重载:定时任务配置支持运行时修改,通过
/api/schedule/reload 接口热重载。
📅 2026-07-27
AI
LLM
DeepSeek
AI 集成方案:从事项分析到报告生成
概述
工作台集成了 AI 能力,覆盖多个业务场景:事项描述质量分析、日报审阅、周报生成、会议纪要整理等。AI 服务的核心是 LLM(大语言模型),通过 DeepSeek API 提供对话能力。
场景一:工作事项描述质量分析
每次扫描 DevOps 新增事项时,AI 自动分析描述是否足够清晰:
def analyze_work_item(title: str, description: str) -> dict:
prompt = f"""分析以下工作事项的描述质量(V4.0标准):
标题:{title}
描述:{description}
请评估:
1. is_clear: 描述是否清晰(布尔值)
2. score: 质量评分 1-10
3. grade: 评级(优秀/良好/一般/较差)
4. reason: 不清晰的原因
5. suggestions: 优化建议
6. missing_items: 缺失的关键信息项"""
response = openai.ChatCompletion.create(
model=LLM_CONFIG["model"],
messages=[{"role": "user", "content": prompt}],
temperature=0.1, # 低温度确保稳定性
)
return parse_result(response)
场景二:日报自动点评
定时任务扫描日报后,AI 自动生成点评意见:
def generate_daily_review(member_name: str, content: str) -> str:
prompt = f"请对组员 {member_name} 的日报进行简洁点评(50字以内):\n{content[:500]}"
# 调用 LLM 获取点评
return review_text
场景三:周报生成
结合本周所有组员的日报内容,AI 自动汇总生成周报草稿:
def summarize_daily_reports(text: str) -> str:
"""根据一周日报汇总生成周报摘要"""
prompt = f"""根据以下一周的组员日报,生成一份简洁的周报汇总:
{text[:10000]}
要求:
1. 分团队工作汇总、重点任务、下周计划三部分
2. 突出关键进展和风险
3. 语言简洁明了"""
return llm_call(prompt)
会议纪要 AI 生成
用户输入会议录音/笔记的原始内容后,AI 自动生成正式会议纪要:
def generate_minutes(raw_content: str) -> str:
prompt = f"""请根据以下会议记录,生成正式的会议纪要:
{raw_content}
会议纪要格式:
一、会议基本信息
二、参会人员
三、会议内容
四、决议事项
五、下一步工作"""
return llm_call(prompt)
配置管理
LLM 配置通过 system_config 表持久化,支持运行时切换模型:
LLM_CONFIG = {
"api_key": os.environ.get("DEEPSEEK_API_KEY", ""),
"base_url": os.environ.get("LLM_BASE_URL", "https://api.deepseek.com"),
"model": os.environ.get("LLM_MODEL", "deepseek-chat"),
}
# 启动时从数据库加载覆盖
db_key = ConfigService.get("llm.api_key", "")
if db_key:
LLM_CONFIG["api_key"] = db_key
LLM_CONFIG["base_url"] = ConfigService.get("llm.base_url", ...)
LLM_CONFIG["model"] = ConfigService.get("llm.model", ...)
经验总结
- 低温度:分析类任务使用 0.1~0.3 的低温度,确保输出稳定性。
- 降级处理:AI 服务不可用时,系统降级使用模板或默认值,核心功能不受影响。
- Token 控制:限制输入长度(如 5000~10000 字符),控制成本并避免超过上下文窗口。
- 配置可切换:支持在设置页随时切换模型(DeepSeek、GPT-4o-mini 等)。