Files
2026-09-26 01:12:10 +08:00

329 lines
13 KiB
Python

import json
from pathlib import Path
import httpx
from django.conf import settings
from django.http import FileResponse, JsonResponse
from django.utils import timezone
from .analysis import METRICS, context, dimensions
from .formula_columns import available_metrics, ensure_available_metrics, validate_column
from .models import FormulaDefinition, FormulaExport
from .views import APIError, audit, body, dataset, endpoint, limit
def design_column(messages, ds, ctx):
if not (settings.AI_ENABLED and settings.OPENAI_API_KEY and settings.OPENAI_MODEL):
raise APIError("尚未配置外部 AI 服务", 503)
if not isinstance(messages, list) or not 1 <= len(messages) <= 12:
raise APIError("对话消息格式无效")
if any(
not isinstance(item, dict)
or set(item) != {"role", "content"}
or item["role"] not in ("user", "assistant")
or not isinstance(item["content"], str)
or len(item["content"]) > 1000
for item in messages
):
raise APIError("对话消息内容无效")
if (
messages[0]["role"] != "user"
or messages[-1]["role"] != "user"
or any(left["role"] == right["role"] for left, right in zip(messages, messages[1:]))
):
raise APIError("对话消息必须按用户描述、助手追问、用户补充的顺序提交")
available = available_metrics(ds)
def expression_schema(depth=0):
leaf = [
{
"type": "object",
"properties": {"metric": {"type": "string", "enum": list(available)}},
"required": ["metric"],
"additionalProperties": False,
},
{
"type": "object",
"properties": {
"number": {"type": "number", "minimum": -1000000000000, "maximum": 1000000000000}
},
"required": ["number"],
"additionalProperties": False,
},
]
if depth >= 8:
return {"anyOf": leaf}
return {
"anyOf": [
*leaf,
{
"type": "object",
"properties": {
"operator": {"type": "string", "enum": ["add", "subtract", "multiply", "divide"]},
"left": expression_schema(depth + 1),
"right": expression_schema(depth + 1),
},
"required": ["operator", "left", "right"],
"additionalProperties": False,
},
]
}
schema = {
"type": "object",
"properties": {
"complete": {"type": "boolean"},
"questions": {"type": "array", "items": {"type": "string"}},
"column": {
"anyOf": [
{"type": "null"},
{
"type": "object",
"properties": {
"name": {"type": "string"},
"meaning": {"type": "string"},
"scope": {"type": "string"},
"missing": {"type": "string", "enum": ["结果不可用(#N/A);零分母不可用"]},
"expression": expression_schema(),
},
"required": ["name", "meaning", "scope", "missing", "expression"],
"additionalProperties": False,
},
]
},
},
"required": ["complete", "questions", "column"],
"additionalProperties": False,
}
system = (
"你是财务计算定义助手。只可用给定指标生成表达式树,不可输出公式文本、SQL或新字段。"
"available_metrics 是当前数据集可计算指标;其他指标依赖字段未映射,必须说明并追问,不可猜测或替代。"
"必须确认列名、含义、适用范围和缺失值/零分母处理。缺少任何信息时 complete=false、column=null,提出具体中文问题。"
"指标节点{metric:指标key},常数{number:数值},运算{operator:add|subtract|multiply|divide,left:...,right:...}。"
"本系统当前仅支持结果不可用(Excel #N/A)且零分母不可用,不支持按零替代或跳过缺失项。"
"用户补充信息后结合整段对话重新判断,只有公式含义无歧义且能表达时 complete=true。"
+ json.dumps(
{
"metrics": METRICS,
"available_metrics": available,
"unavailable_metrics": {key: value for key, value in METRICS.items() if key not in available},
"dimensions": dimensions(ds),
"context": {k: ctx[k] for k in ("current", "baseline", "filters", "group")},
},
ensure_ascii=False,
)
)
payload = {
"model": settings.OPENAI_MODEL,
"store": False,
"max_output_tokens": 1400,
"input": [{"role": "system", "content": system}, *messages],
"text": {
"format": {
"type": "json_schema",
"name": "formula_column_design",
"strict": True,
"schema": schema,
}
},
}
try:
with httpx.Client(timeout=30) as client:
response = client.post(
"https://api.openai.com/v1/responses",
headers={"Authorization": f"Bearer {settings.OPENAI_API_KEY}"},
json=payload,
)
response.raise_for_status()
result = response.json()
if not isinstance(result, dict) or result.get("status") != "completed":
raise ValueError("incomplete response")
text = next(c["text"] for o in result["output"] for c in o["content"] if c["type"] == "output_text")
design = json.loads(text)
if (
set(design) != {"complete", "questions", "column"}
or type(design["complete"]) is not bool
or not isinstance(design["questions"], list)
or len(design["questions"]) > 8
):
raise ValueError("invalid response")
if design["complete"]:
if design["column"] is None:
raise ValueError("complete response has no column")
design["column"] = validate_column(design["column"])
try:
ensure_available_metrics(ds, design["column"])
except ValueError as exc:
return {"complete": False, "questions": [str(exc)], "column": None}
if design["questions"]:
raise ValueError("complete response contains questions")
elif (
design["column"] is not None
or not design["questions"]
or any(not isinstance(q, str) or not q.strip() or len(q) > 500 for q in design["questions"])
):
raise ValueError("incomplete response malformed")
return design
except (httpx.HTTPError, ValueError, KeyError, TypeError, StopIteration) as exc:
raise APIError("模型暂时不可用或返回了不安全的计算定义;未创建工作簿", 502) from exc
@endpoint(("POST",), roles=("admin", "analyst"))
def formula_design(request, dataset_id):
ds = dataset(request, dataset_id, True)
data = body(request)
if request.member.departments:
raise APIError("科室授权账号不可定义跨范围公式导出", 403)
ctx = context(ds, data.get("context", {}))
if data.get("consent") is not True:
raise APIError("请先同意发送列描述和筛选范围到外部 AI", 400)
limit(f"formula-design:{request.user.pk}", 20, 60)
result = design_column(data.get("messages"), ds, ctx)
audit(request, "formula_export.design", ds.pk, {"complete": result["complete"]})
return JsonResponse(result)
@endpoint(("GET", "POST"), roles=("admin", "analyst"))
def formula_definitions(request, dataset_id):
ds = dataset(request, dataset_id, True)
if request.member.departments:
raise APIError("科室授权账号不可管理跨范围公式定义", 403)
if request.method == "GET":
items = FormulaDefinition.objects.filter(dataset=ds, user=request.user).order_by("normalized_name")
return JsonResponse(
{
"items": [
{
"id": str(item.pk),
"definition": item.definition,
"updated_at": item.updated_at.isoformat(),
}
for item in items
]
}
)
definition = validate_column(body(request).get("definition"))
try:
ensure_available_metrics(ds, definition)
except ValueError as exc:
raise APIError(str(exc)) from exc
normalized_name = definition["name"].strip().casefold()
existing = FormulaDefinition.objects.filter(
dataset=ds, user=request.user, normalized_name=normalized_name
).first()
if not existing and FormulaDefinition.objects.filter(dataset=ds, user=request.user).count() >= 50:
raise APIError("每个数据集最多保存 50 个公式定义")
item, created = FormulaDefinition.objects.update_or_create(
dataset=ds,
user=request.user,
normalized_name=normalized_name,
defaults={"definition": definition},
)
audit(request, "formula_definition.save", item.pk, {"dataset": str(ds.pk), "name": definition["name"]})
return JsonResponse(
{
"id": str(item.pk),
"definition": item.definition,
"updated_at": item.updated_at.isoformat(),
},
status=201 if created else 200,
)
@endpoint(("DELETE",), roles=("admin", "analyst"))
def formula_definition_delete(request, dataset_id, definition_id):
ds = dataset(request, dataset_id, True)
if request.member.departments:
raise APIError("科室授权账号不可管理跨范围公式定义", 403)
item = FormulaDefinition.objects.filter(pk=definition_id, dataset=ds, user=request.user).first()
if not item:
raise APIError("公式定义不存在", 404)
name = item.definition.get("name", "")
audit(request, "formula_definition.delete", item.pk, {"dataset": str(ds.pk), "name": name})
item.delete()
return JsonResponse({"deleted": True})
@endpoint(("POST",), roles=("admin", "analyst"))
def formula_export_create(request, dataset_id):
ds = dataset(request, dataset_id, True)
if request.member.departments:
raise APIError("科室授权账号不可创建公式导出", 403)
data = body(request)
ctx = context(ds, data)
definitions = data.get("custom_columns", [])
if not isinstance(definitions, list) or len(definitions) > 10:
raise APIError("自定义列最多 10 个")
for item in definitions:
validate_column(item)
try:
ensure_available_metrics(ds, item)
except ValueError as exc:
raise APIError(str(exc)) from exc
limit(f"formula-export:{request.user.pk}", 5, 3600)
job = FormulaExport.objects.create(
dataset=ds,
user=request.user,
context={**ctx, "custom_columns": definitions},
departments=list(request.member.departments),
)
audit(
request,
"formula_export.create",
job.pk,
{"dataset": str(ds.pk), "columns": [x["name"] for x in definitions]},
)
return JsonResponse({"id": str(job.pk), "status": job.status}, status=202)
def _owned_job(request, dataset_id, export_id):
job = FormulaExport.objects.filter(
pk=export_id,
dataset_id=dataset_id,
dataset__organization=request.member.organization,
user=request.user,
).first()
if not job:
raise APIError("导出任务不存在", 404)
if set(request.member.departments) != set(job.departments):
raise APIError("授权范围已变化,不能访问此导出", 403)
return job
@endpoint(("GET",))
def formula_export_status(request, dataset_id, export_id):
job = _owned_job(request, dataset_id, export_id)
return JsonResponse(
{
"id": str(job.pk),
"status": job.status,
"error": "生成失败,请联系管理员并提供任务编号" if job.status == "failed" else "",
"expires_at": job.expires_at.isoformat() if job.expires_at else None,
}
)
@endpoint(("GET",))
def formula_export_download(request, dataset_id, export_id):
job = _owned_job(request, dataset_id, export_id)
if job.status != "succeeded" or not job.path or not job.expires_at or job.expires_at <= timezone.now():
raise APIError("文件尚未就绪或已过期", 410 if job.status == "succeeded" else 409)
export_root = Path(settings.MEDIA_ROOT) / "formula-exports"
candidate = Path(job.path)
expected = export_root / str(job.user_id) / f"{job.pk}.xlsx"
root = export_root.resolve()
path = candidate.resolve()
if (
export_root.is_symlink()
or candidate.is_symlink()
or candidate.parent.is_symlink()
or path != expected.resolve()
or not path.is_relative_to(root)
or not path.is_file()
):
raise APIError("导出文件不存在", 410)
audit(request, "formula_export.download", job.pk)
return FileResponse(path.open("rb"), as_attachment=True, filename="traceiq-formula-analysis.xlsx")