"""知识摘要 API — 管理会计OS持久记忆 提供: - 查询最近摘要列表 - 查询单个摘要详情 - 手动触发各层级摘要生成 - 查询未摘要的事件 """ from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.orm import Session from sqlalchemy import func, desc from datetime import datetime, timedelta from typing import Optional from app.database import get_db from app.auth_middleware import require_role from app.models import KnowledgeEvent, KnowledgeSummary from app.services.knowledge_service import ( generate_summary_sync, generate_daily_sync, generate_weekly_sync, generate_monthly_sync, get_last_summary, extract_events, ) import logging logger = logging.getLogger("cma.knowledge_api") router = APIRouter(prefix="/api/cma/knowledge", tags=["知识摘要"], dependencies=[Depends(require_role("ceo", "finance", "business", "it"))], ) def summary_to_dict(s: KnowledgeSummary) -> dict: return { "id": s.id, "level": s.level, "period_key": s.period_key, "title": s.title, "content": s.content, "kpi_changes": s.kpi_changes, "decision_points": s.decision_points, "key_metrics": s.key_metrics, "prev_summary_id": s.prev_summary_id, "model": s.model, "is_stale": s.is_stale, "created_at": s.created_at.isoformat() if s.created_at else None, } # ── 查询 ── @router.get("/summaries") def list_summaries( level: Optional[str] = None, limit: int = 20, offset: int = 0, db: Session = Depends(get_db), ): """获取摘要列表，按层级筛选，按时间倒序""" query = db.query(KnowledgeSummary) if level: query = query.filter(KnowledgeSummary.level == level) query = query.order_by(desc(KnowledgeSummary.id)).offset(offset).limit(limit) total = db.query(func.count(KnowledgeSummary.id)).select_from(KnowledgeSummary) if level: total = total.filter(KnowledgeSummary.level == level) total = total.scalar() return { "total": total, "items": [summary_to_dict(s) for s in query.all()], } @router.get("/summaries/latest") def latest_summary( level: str = Query("daily", description="层级: daily/weekly/monthly/cumulative"), db: Session = Depends(get_db), ): """获取指定层级的最新摘要""" s = get_last_summary(db, level) if not s: return {"detail": f"没有{level}层级的摘要"}, 404 return summary_to_dict(s) @router.get("/summaries/{summary_id}") def get_summary(summary_id: int, db: Session = Depends(get_db)): """获取单条摘要详情""" s = db.query(KnowledgeSummary).filter(KnowledgeSummary.id == summary_id).first() if not s: raise HTTPException(status_code=404, detail="摘要不存在") return summary_to_dict(s) # ── 事件查询 ── @router.get("/events") def list_events( since: Optional[str] = None, until: Optional[str] = None, limit: int = 50, db: Session = Depends(get_db), ): """查询未摘要的原始事件 如果不传时间，默认返回最近7天的操作记录和预警。 """ try: dt_since = datetime.fromisoformat(since) if since else datetime.utcnow() - timedelta(days=7) dt_until = datetime.fromisoformat(until) if until else datetime.utcnow() except ValueError: raise HTTPException(status_code=400, detail="时间格式错误，请使用 ISO 格式如 2026-06-01T00:00:00") events = extract_events(db, dt_since, dt_until) return {"since": dt_since.isoformat(), "until": dt_until.isoformat(), "total": len(events), "events": events[:limit]} # ── 手动触发 ── @router.post("/generate/daily") def trigger_daily_summary(db: Session = Depends(get_db)): """手动触发每日摘要生成""" try: result = generate_daily_sync(db) return {"message": "每日摘要已生成", "summary": result} except Exception as e: logger.exception("每日摘要生成失败") raise HTTPException(status_code=500, detail=f"生成失败: {str(e)}") @router.post("/generate/weekly") def trigger_weekly_summary(db: Session = Depends(get_db)): """手动触发周度摘要生成""" try: result = generate_weekly_sync(db) return {"message": "周度摘要已生成", "summary": result} except Exception as e: logger.exception("周度摘要生成失败") raise HTTPException(status_code=500, detail=f"生成失败: {str(e)}") @router.post("/generate/monthly") def trigger_monthly_summary(db: Session = Depends(get_db)): """手动触发月度摘要生成""" try: result = generate_monthly_sync(db) return {"message": "月度摘要已生成", "summary": result} except Exception as e: logger.exception("月度摘要生成失败") raise HTTPException(status_code=500, detail=f"生成失败: {str(e)}") # ── CMA P1+P2 Round 3: 职业道德自测题 ── ETHICS_QUIZ_DATA = [ { "id": 1, "scenario": '你发现公司财务数据异常（现金比率仅0.6%），但总经理说"先别声张，等融资款到账再说"，你应该：', "options": [ {"id": "a", "text": "听从老板，暂时隐瞒", "correct": False, "feedback": "违反IMA诚信准则。财务人员有义务如实记录和报告"}, {"id": "b", "text": "如实记录异常并向上级或审计委员会报告", "correct": True, "feedback": "诚信+客观准则。IMA要求财务人员如实反映财务状况"}, {"id": "c", "text": "立即辞职，一走了之", "correct": False, "feedback": "可行但非最优解。应先通过正规渠道报告问题"}, ], }, { "id": 2, "scenario": "竞争对手公司的高管在行业会议上向你打探你们公司的成本数据，你应该：", "options": [ {"id": "a", "text": "告知对方，反正行业数据都差不多", "correct": False, "feedback": "违反保密准则。成本数据属于公司机密"}, {"id": "b", "text": "断然拒绝，并告知这是公司机密", "correct": True, "feedback": "保密准则。财务人员必须保护公司机密信息"}, {"id": "c", "text": "用编造的假数据糊弄对方", "correct": False, "feedback": "违反诚信准则。即使对竞争对手也不应提供虚假信息"}, ], }, { "id": 3, "scenario": "你被要求分析一项复杂的金融衍生品对公司的影响，但你没有相关专业知识经验，应该：", "options": [ {"id": "a", "text": "凭感觉估算一个数值交差", "correct": False, "feedback": "违反胜任准则。不能在自己不擅长的领域出具专业意见"}, {"id": "b", "text": "如实告知自己缺乏经验，建议聘请外部专家", "correct": True, "feedback": "胜任准则。IMA要求持续学习并在能力范围内执业"}, {"id": "c", "text": "花一个晚上学习后立即出报告", "correct": False, "feedback": "短期内突击学习难以达到专业标准，建议请专家"}, ], }, { "id": 4, "scenario": "供应商送你价值5000元的高档礼品，并暗示希望在招标中获得倾斜，你应该：", "options": [ {"id": "a", "text": "收下礼物，在招标中适当照顾", "correct": False, "feedback": "违反客观准则。礼品影响职业判断的独立性"}, {"id": "b", "text": "拒收礼品并报告公司合规部门", "correct": True, "feedback": "客观准则。财务人员必须保持独立性和客观判断"}, {"id": "c", "text": "收下但不做任何特殊照顾", "correct": False, "feedback": "即使不特殊照顾，收礼本身已违反公司廉洁规定"}, ], }, { "id": 5, "scenario": '销售总监要求你将下季度已签约但未发货的订单提前确认为本季度收入，以"完成业绩指标"，你应该：', "options": [ {"id": "a", "text": "照做，反正迟早会确认", "correct": False, "feedback": "违反诚信准则。收入确认必须符合会计准则的确认条件"}, {"id": "b", "text": "拒绝并解释收入确认应遵循权责发生制", "correct": True, "feedback": "诚信+客观准则。IMA要求如实反映交易实质"}, {"id": "c", "text": "折中处理，先确认一半", "correct": False, "feedback": "部分确认依然违反会计准则和IMA诚信准则"}, ], }, { "id": 6, "scenario": "你的亲戚是股票投资者，向你打听公司即将发布的季度盈利数据（尚未公开），你应该：", "options": [ {"id": "a", "text": "暗示性地透露一点，让亲戚自己去判断", "correct": False, "feedback": "违反保密准则。内幕信息在任何情况下都不能泄露"}, {"id": "b", "text": "明确拒绝并告知这是未公开的内部信息", "correct": True, "feedback": "保密准则。IMA要求保护信息机密性，禁止内幕交易"}, {"id": "c", "text": "告知所有数据需等正式发布后才能讨论", "correct": False, "feedback": "虽然拒绝了但未明确说明信息性质，应更直接拒绝"}, ], }, { "id": 7, "scenario": "你在编制税务筹划方案时，发现最新的税法已经更新，但你还没参加过相关培训，应该：", "options": [ {"id": "a", "text": "按老法规继续编制，客户不会发现", "correct": False, "feedback": "违反胜任准则。必须掌握最新知识才能提供专业服务"}, {"id": "b", "text": "先学习新税法，再基于最新法规编制方案", "correct": True, "feedback": "胜任准则。IMA要求持续提升专业能力，保持知识更新"}, {"id": "c", "text": "请客户另请高明", "correct": False, "feedback": "回避问题不可取。应先通过学习解决问题"}, ], }, { "id": 8, "scenario": "月度经营分析会上，CEO要求你只汇报好的指标，不提及存在的风险问题，你应该：", "options": [ {"id": "a", "text": "听从CEO安排，只报喜不报忧", "correct": False, "feedback": "违反客观准则。选择性披露误导决策"}, {"id": "b", "text": "完整披露所有指标，包括存在风险和问题", "correct": True, "feedback": "客观+诚信准则。IMA要求全面、公正地呈现信息"}, {"id": "c", "text": "在会上只口头上提一下风险但不出现在书面报告里", "correct": False, "feedback": "口头提及无记录，核心问题仍需书面报告"}, ], }, { "id": 9, "scenario": "为达成KPI目标获得年终奖，部门经理要求你修改部分实际运营数据，你应该：", "options": [ {"id": "a", "text": "修改数据，反正金额不大", "correct": False, "feedback": "违反诚信准则。任何金额的数据造假都不可接受"}, {"id": "b", "text": "坚决拒绝并向上级或审计委员会举报", "correct": True, "feedback": "诚信准则。诚实是财务人员的首要职业道德要求"}, {"id": "c", "text": "假装答应但不执行", "correct": False, "feedback": "不明确的回应可能导致后续施压，应明确拒绝"}, ], }, { "id": 10, "scenario": "你准备离职去竞争对手公司，离职前应该如何处理手头的公司财务数据文件？", "options": [ {"id": "a", "text": "拷贝一份带走，作为新公司的参考", "correct": False, "feedback": "违反保密准则。离职不得带走公司机密数据"}, {"id": "b", "text": "全部归还公司，并确认不保留任何副本", "correct": True, "feedback": "保密准则。离职后仍须承担保密义务"}, {"id": "c", "text": "只带走非敏感性的一般统计资料", "correct": False, "feedback": "任何公司未公开数据均不得外传"}, ], }, ] @router.get("/ethics-quiz") def get_ethics_quiz(): """返回10道IMA职业道德情景自测题""" return ETHICS_QUIZ_DATA