611 lines
22 KiB
Python
611 lines
22 KiB
Python
import csv
|
|
import hashlib
|
|
import io
|
|
import ipaddress
|
|
import json
|
|
import math
|
|
import tempfile
|
|
import uuid
|
|
from datetime import timedelta
|
|
from functools import wraps
|
|
from pathlib import Path
|
|
|
|
from django.conf import settings
|
|
from django.contrib.auth import authenticate, login, logout
|
|
from django.contrib.auth.models import User
|
|
from django.contrib.auth.password_validation import validate_password
|
|
from django.core.exceptions import ValidationError
|
|
from django.db import IntegrityError, connection, transaction
|
|
from django.http import FileResponse, HttpResponse, JsonResponse
|
|
from django.middleware.csrf import get_token
|
|
from django.utils import timezone
|
|
from openpyxl import Workbook
|
|
from openpyxl.cell import WriteOnlyCell
|
|
|
|
from .analysis import METRICS, analyze, context, default_preferences, dimensions, scoped_facts
|
|
from .importing import file_digest, profile_file, suggest_mapping, validate_mapping
|
|
from .models import Audit, Dataset, Investigation, Membership, Preference, Throttle
|
|
|
|
|
|
class APIError(Exception):
|
|
def __init__(self, message, status=400):
|
|
self.message, self.status = message, status
|
|
|
|
|
|
def endpoint(methods=("GET",), roles=None, public=False):
|
|
def decorate(fn):
|
|
@wraps(fn)
|
|
def view(request, *args, **kwargs):
|
|
try:
|
|
if request.method not in methods:
|
|
raise APIError("请求方法不允许", 405)
|
|
if not public:
|
|
if not request.user.is_authenticated:
|
|
raise APIError("请先登录", 401)
|
|
try:
|
|
request.member = request.user.membership
|
|
except Membership.DoesNotExist:
|
|
raise APIError("账号尚未分配组织", 403)
|
|
if roles and request.member.role not in roles:
|
|
raise APIError("没有此操作的权限", 403)
|
|
return fn(request, *args, **kwargs)
|
|
except APIError as exc:
|
|
return JsonResponse({"error": exc.message}, status=exc.status)
|
|
except (ValueError, ValidationError, json.JSONDecodeError) as exc:
|
|
return JsonResponse({"error": str(exc)}, status=400)
|
|
|
|
return view
|
|
|
|
return decorate
|
|
|
|
|
|
def body(request):
|
|
if request.content_type != "application/json":
|
|
raise APIError("请求必须使用 application/json")
|
|
value = json.loads(request.body or b"{}")
|
|
if not isinstance(value, dict):
|
|
raise APIError("请求必须是 JSON 对象")
|
|
return value
|
|
|
|
|
|
def audit(request, action, resource="", details=None):
|
|
Audit.objects.create(
|
|
organization=request.member.organization,
|
|
user=request.user,
|
|
action=action,
|
|
resource=str(resource),
|
|
details=details or {},
|
|
)
|
|
|
|
|
|
def limit(key, count=10, seconds=900):
|
|
key = hashlib.sha256(key.encode()).hexdigest()
|
|
now = timezone.now()
|
|
with transaction.atomic():
|
|
item, _ = Throttle.objects.get_or_create(
|
|
key=key, defaults={"expires_at": now + timedelta(seconds=seconds)}
|
|
)
|
|
item = Throttle.objects.select_for_update().get(pk=key)
|
|
if item.expires_at <= now:
|
|
item.count, item.expires_at = 0, now + timedelta(seconds=seconds)
|
|
if item.count >= count:
|
|
raise APIError("请求过于频繁，请稍后重试", 429)
|
|
item.count += 1
|
|
item.save()
|
|
|
|
|
|
def dataset(request, dataset_id, ready=False):
|
|
ds = Dataset.objects.filter(pk=dataset_id, organization=request.member.organization).first()
|
|
if not ds:
|
|
raise APIError("数据集不存在", 404)
|
|
if ready and ds.status != "ready":
|
|
raise APIError("数据集尚未导入完成", 409)
|
|
return ds
|
|
|
|
|
|
def preferences(request, ds):
|
|
pref = Preference.objects.filter(dataset=ds, user=request.user).first()
|
|
return {**default_preferences(ds), **(pref.value if pref else {})}, pref.revision if pref else 0
|
|
|
|
|
|
def serialize_dataset(ds, detailed=False, restricted=False, raw_access=True):
|
|
out = {
|
|
"id": str(ds.pk),
|
|
"name": ds.name,
|
|
"status": ds.status,
|
|
"version": ds.sha256,
|
|
"row_count": None if restricted else ds.row_count,
|
|
"periods": ds.periods,
|
|
"created_at": ds.created_at.isoformat(),
|
|
"error": ds.error if not restricted else "",
|
|
"quality": ds.quality if not restricted else {},
|
|
"dimensions": dimensions(ds) if ds.mapping else [],
|
|
}
|
|
if detailed:
|
|
out.update(mapping=ds.mapping)
|
|
if not restricted and raw_access:
|
|
out["profile"] = ds.profile
|
|
return out
|
|
|
|
|
|
@endpoint(public=True)
|
|
def session(request):
|
|
data = {
|
|
"csrf": get_token(request),
|
|
"user": None,
|
|
"ai_enabled": settings.AI_ENABLED and bool(settings.OPENAI_API_KEY and settings.OPENAI_MODEL),
|
|
}
|
|
if request.user.is_authenticated:
|
|
try:
|
|
m = request.user.membership
|
|
data["user"] = {
|
|
"id": request.user.pk,
|
|
"name": request.user.username,
|
|
"role": m.role,
|
|
"organization": m.organization.name,
|
|
"departments": m.departments,
|
|
}
|
|
except Membership.DoesNotExist:
|
|
pass
|
|
return JsonResponse(data)
|
|
|
|
|
|
@endpoint(("POST",), public=True)
|
|
def sign_in(request):
|
|
data = body(request)
|
|
username, password = data.get("username", ""), data.get("password", "")
|
|
if (
|
|
not isinstance(username, str)
|
|
or not isinstance(password, str)
|
|
or len(username) > 150
|
|
or len(password) > 1024
|
|
):
|
|
raise APIError("登录参数无效")
|
|
address = request.META.get("REMOTE_ADDR", "unknown")
|
|
if settings.TRUST_PROXY and request.META.get("HTTP_X_REAL_IP"):
|
|
try:
|
|
address = str(ipaddress.ip_address(request.META["HTTP_X_REAL_IP"]))
|
|
except ValueError:
|
|
raise APIError("代理来源地址无效")
|
|
limit("login-ip:" + address, 30)
|
|
limit("login-user:" + username.casefold(), 10)
|
|
user = authenticate(request, username=username, password=password)
|
|
if not user or not Membership.objects.filter(user=user).exists():
|
|
raise APIError("用户名或密码错误", 401)
|
|
login(request, user)
|
|
request.member = user.membership
|
|
audit(request, "auth.login")
|
|
return JsonResponse({"ok": True, "csrf": get_token(request)})
|
|
|
|
|
|
@endpoint(("POST",))
|
|
def sign_out(request):
|
|
audit(request, "auth.logout")
|
|
logout(request)
|
|
return JsonResponse({"ok": True})
|
|
|
|
|
|
@endpoint(("GET", "POST"))
|
|
def datasets(request):
|
|
if request.method == "GET":
|
|
return JsonResponse(
|
|
{
|
|
"items": [
|
|
serialize_dataset(ds, restricted=bool(request.member.departments))
|
|
for ds in Dataset.objects.filter(organization=request.member.organization).order_by(
|
|
"-created_at"
|
|
)[:100]
|
|
]
|
|
}
|
|
)
|
|
if request.member.role not in ("admin", "analyst") or request.member.departments:
|
|
raise APIError("仅有完整数据权限的管理员或分析师可导入", 403)
|
|
limit(f"upload:{request.user.pk}", 20, 3600)
|
|
file = request.FILES.get("file")
|
|
if not file or not 0 < file.size <= settings.MAX_UPLOAD_BYTES:
|
|
raise APIError("文件为空或超过上传大小限制")
|
|
ext = Path(file.name).suffix.lower()
|
|
if ext not in (".xlsx", ".csv"):
|
|
raise APIError("仅接受 .xlsx 或 UTF-8 CSV")
|
|
digest = file_digest(file)
|
|
existing = Dataset.objects.filter(organization=request.member.organization, sha256=digest).first()
|
|
if existing:
|
|
return JsonResponse(
|
|
serialize_dataset(
|
|
existing,
|
|
True,
|
|
raw_access=request.member.role == "admin"
|
|
or (existing.created_by_id == request.user.pk and existing.status in ("mapping", "failed")),
|
|
),
|
|
status=200,
|
|
)
|
|
ds = Dataset(
|
|
organization=request.member.organization,
|
|
created_by=request.user,
|
|
name=Path(file.name).name[:200],
|
|
sha256=digest,
|
|
)
|
|
ds.file.save(str(uuid.uuid4()) + ext, file, save=False)
|
|
try:
|
|
ds.profile = profile_file(ds.file.path)
|
|
ds.mapping = suggest_mapping(ds.profile)
|
|
ds.save()
|
|
except IntegrityError:
|
|
ds.file.delete(save=False)
|
|
return JsonResponse(
|
|
serialize_dataset(
|
|
Dataset.objects.get(organization=request.member.organization, sha256=digest),
|
|
True,
|
|
raw_access=request.member.role == "admin",
|
|
)
|
|
)
|
|
except Exception as exc:
|
|
ds.file.delete(save=False)
|
|
raise APIError(
|
|
"无法读取文件，请检查格式或文件安全限制："
|
|
+ (str(exc)[:160] if isinstance(exc, (ValueError, UnicodeError)) else "文件损坏")
|
|
) from exc
|
|
audit(request, "dataset.upload", ds.pk, {"sha256": digest})
|
|
return JsonResponse(serialize_dataset(ds, True), status=201)
|
|
|
|
|
|
@endpoint()
|
|
def dataset_detail(request, dataset_id):
|
|
ds = dataset(request, dataset_id)
|
|
raw_access = request.member.role == "admin" or (
|
|
ds.created_by_id == request.user.pk and ds.status in ("mapping", "failed")
|
|
)
|
|
result = serialize_dataset(ds, True, bool(request.member.departments), raw_access)
|
|
if ds.status == "ready":
|
|
result["preferences"], result["revision"] = preferences(request, ds)
|
|
return JsonResponse(result)
|
|
|
|
|
|
@endpoint(("POST",), roles=("admin", "analyst"))
|
|
def mapping(request, dataset_id):
|
|
ds = dataset(request, dataset_id)
|
|
if request.member.departments:
|
|
raise APIError("没有修改导入映射的权限", 403)
|
|
data = body(request)
|
|
if data.get("suggest"):
|
|
return JsonResponse(suggest_mapping(ds.profile, data.get("sheet"), data.get("header")))
|
|
value = validate_mapping(data, ds.profile)
|
|
updated = Dataset.objects.filter(pk=ds.pk, status__in=["mapping", "failed"]).update(
|
|
mapping=value, status="queued", error="", started_at=None, finished_at=None
|
|
)
|
|
if not updated:
|
|
raise APIError("数据版本已锁定，不能重复导入或修改", 409)
|
|
audit(request, "dataset.confirm_mapping", ds.pk, {"mapping": value})
|
|
return JsonResponse({"status": "queued"}, status=202)
|
|
|
|
|
|
@endpoint(("POST",))
|
|
def query(request, dataset_id):
|
|
ds = dataset(request, dataset_id, True)
|
|
pref, _ = preferences(request, ds)
|
|
limit(f"analysis:{request.user.pk}", 120, 60)
|
|
return JsonResponse(analyze(ds, request.member, body(request), pref))
|
|
|
|
|
|
@endpoint(("POST",))
|
|
def evidence(request, dataset_id):
|
|
ds = dataset(request, dataset_id, True)
|
|
data = body(request)
|
|
ctx = context(ds, data)
|
|
offset, size = data.get("offset", 0), data.get("limit", 50)
|
|
if type(offset) is not int or type(size) is not int or not 0 <= offset <= 1000000 or not 1 <= size <= 200:
|
|
raise APIError("分页参数无效")
|
|
side = data.get("side", "current")
|
|
if side not in ("current", "baseline"):
|
|
raise APIError("期间选择无效")
|
|
qs = scoped_facts(ds, request.member, ctx["filters"]).filter(period__in=ctx[side]).order_by("source_row")
|
|
rows = list(
|
|
qs[offset : offset + size].values(
|
|
"source_row", "period", "dimensions", "revenue", "cost", "gross", "raw"
|
|
)
|
|
)
|
|
# Raw patient/customer identifiers require an explicit administrator role.
|
|
if request.member.role != "admin":
|
|
for row in rows:
|
|
row.pop("raw")
|
|
audit(
|
|
request,
|
|
"evidence.read",
|
|
ds.pk,
|
|
{"context": ctx, "side": side, "offset": offset, "limit": size},
|
|
)
|
|
return JsonResponse(
|
|
{
|
|
"rows": rows,
|
|
"total": qs.count(),
|
|
"context": ctx,
|
|
"version": ds.sha256,
|
|
"sheet": ds.mapping["sheet"],
|
|
}
|
|
)
|
|
|
|
|
|
def csv_safe(value):
|
|
text = str(value) if value is not None else ""
|
|
if not isinstance(value, str):
|
|
return text
|
|
return (
|
|
"'" + text
|
|
if text.startswith(("\t", "\r", "\n")) or text.lstrip().startswith(("=", "+", "-", "@"))
|
|
else text
|
|
)
|
|
|
|
|
|
@endpoint(("POST",), roles=("admin", "analyst"))
|
|
def export(request, dataset_id):
|
|
ds = dataset(request, dataset_id, True)
|
|
data = body(request)
|
|
ctx = context(ds, data)
|
|
limit(f"export:{request.user.pk}", 10, 3600)
|
|
kind = data.get("format", "csv")
|
|
if kind not in ("csv", "xlsx"):
|
|
raise APIError("导出格式无效")
|
|
qs = (
|
|
scoped_facts(ds, request.member, ctx["filters"])
|
|
.filter(period__in=ctx["current"] + ctx["baseline"])
|
|
.order_by("source_row")
|
|
)
|
|
if qs.count() > 100000:
|
|
raise APIError("单次导出上限 100,000 行，请缩小期间或筛选范围")
|
|
output = tempfile.TemporaryFile(mode="w+b")
|
|
csv_stream = None
|
|
if kind == "csv":
|
|
output.write(b"\xef\xbb\xbf")
|
|
csv_stream = io.TextIOWrapper(output, encoding="utf-8", newline="", write_through=True)
|
|
csv_writer = csv.writer(csv_stream) if csv_stream else None
|
|
workbook = Workbook(write_only=True) if kind == "xlsx" else None
|
|
sheet = workbook.create_sheet("证据明细") if workbook else None
|
|
|
|
def write_row(values):
|
|
if sheet is None:
|
|
csv_writer.writerow(csv_safe(value) for value in values)
|
|
else:
|
|
cells = []
|
|
for value in values:
|
|
# Excel has only 15 significant digits; evidence amounts stay exact text.
|
|
cell = WriteOnlyCell(sheet, value=str(value) if value is not None else "")
|
|
cell.data_type = "s"
|
|
cells.append(cell)
|
|
sheet.append(cells)
|
|
|
|
try:
|
|
write_row(
|
|
[
|
|
"TraceIQ 数据版本",
|
|
ds.sha256,
|
|
]
|
|
)
|
|
write_row(["筛选上下文", json.dumps(ctx, ensure_ascii=False)])
|
|
write_row(["ROI 口径", ds.mapping["roi_formula"]])
|
|
write_row(["导出时间 UTC", timezone.now().isoformat()])
|
|
dims = dimensions(ds)
|
|
write_row(
|
|
[
|
|
"原始工作表",
|
|
"原始行号",
|
|
"期间",
|
|
*[d["name"] for d in dims],
|
|
"实际收入",
|
|
"成本",
|
|
"毛利贡献",
|
|
]
|
|
)
|
|
for row in qs.iterator(chunk_size=1000):
|
|
write_row(
|
|
[
|
|
ds.mapping["sheet"],
|
|
row.source_row,
|
|
row.period,
|
|
*[row.dimensions.get(d["key"], "") for d in dims],
|
|
row.revenue,
|
|
row.cost,
|
|
row.gross,
|
|
]
|
|
)
|
|
if workbook:
|
|
workbook.save(output)
|
|
content_type = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
|
else:
|
|
csv_stream.flush()
|
|
csv_stream.detach()
|
|
csv_stream = None
|
|
content_type = "text/csv; charset=utf-8"
|
|
output.seek(0)
|
|
audit(request, "evidence.export", ds.pk, {"context": ctx, "format": kind})
|
|
return FileResponse(
|
|
output,
|
|
as_attachment=True,
|
|
filename=f"traceiq-evidence.{kind}",
|
|
content_type=content_type,
|
|
)
|
|
except Exception:
|
|
if csv_stream is not None:
|
|
csv_stream.close()
|
|
else:
|
|
output.close()
|
|
raise
|
|
|
|
|
|
@endpoint(("GET", "PUT"))
|
|
def settings_view(request, dataset_id):
|
|
ds = dataset(request, dataset_id, True)
|
|
if request.method == "GET":
|
|
value, revision = preferences(request, ds)
|
|
return JsonResponse({"value": value, "revision": revision})
|
|
data = body(request)
|
|
value = data.get("value")
|
|
defaults = default_preferences(ds)
|
|
if not isinstance(value, dict) or set(value) != set(defaults):
|
|
raise APIError("设置字段不完整")
|
|
allowed = {d["key"] for d in dimensions(ds)}
|
|
hierarchy = value["hierarchy"]
|
|
if (
|
|
not isinstance(hierarchy, list)
|
|
or not hierarchy
|
|
or any(not isinstance(k, str) or k not in allowed for k in hierarchy)
|
|
or len(set(hierarchy)) != len(hierarchy)
|
|
):
|
|
raise APIError("下钻层级无效或重复")
|
|
if (
|
|
value["entry"] not in ("overview", "anomalies")
|
|
or value["comparison"] not in ("yoy", "mom")
|
|
or not isinstance(value["metric"], str)
|
|
or value["metric"] not in METRICS
|
|
):
|
|
raise APIError("默认视图或指标无效")
|
|
for field, lo, hi in [("threshold", 0, 10), ("min_impact", 0, 1e15), ("min_samples", 1, 1000000)]:
|
|
v = value[field]
|
|
if type(v) not in (int, float) or not math.isfinite(v) or not lo <= v <= hi:
|
|
raise APIError("异常阈值无效")
|
|
if type(value["min_samples"]) is not int:
|
|
raise APIError("最小样本数必须为整数")
|
|
for field in ["show_heatmap", "show_queue", "show_copilot", "alerts"]:
|
|
if type(value[field]) is not bool:
|
|
raise APIError("显示设置必须为布尔值")
|
|
with transaction.atomic():
|
|
pref = Preference.objects.select_for_update().filter(dataset=ds, user=request.user).first()
|
|
if data.get("revision") != (pref.revision if pref else 0):
|
|
raise APIError("设置已在其他窗口更新，请重新打开", 409)
|
|
if pref:
|
|
pref.value, pref.revision = value, pref.revision + 1
|
|
pref.save()
|
|
else:
|
|
try:
|
|
pref = Preference.objects.create(dataset=ds, user=request.user, value=value)
|
|
except IntegrityError as exc:
|
|
raise APIError("设置发生并发更新，请重新打开", 409) from exc
|
|
audit(request, "settings.update", ds.pk, {"revision": pref.revision, "value": value})
|
|
return JsonResponse({"value": pref.value, "revision": pref.revision})
|
|
|
|
|
|
@endpoint(("GET", "POST"))
|
|
def investigations(request, dataset_id):
|
|
ds = dataset(request, dataset_id, True)
|
|
if request.method == "GET":
|
|
return JsonResponse(
|
|
{
|
|
"items": list(
|
|
Investigation.objects.filter(dataset=ds, user=request.user).values(
|
|
"id", "key", "context", "status", "note", "updated_at"
|
|
)
|
|
)
|
|
}
|
|
)
|
|
if request.member.role == "viewer":
|
|
raise APIError("查看者不可修改调查状态", 403)
|
|
data = body(request)
|
|
ctx = context(ds, data.get("context", {}))
|
|
status, note = data.get("status"), data.get("note", "")
|
|
if status not in ("open", "reviewing", "resolved") or not isinstance(note, str) or len(note) > 2000:
|
|
raise APIError("调查状态或备注无效")
|
|
key = hashlib.sha256(json.dumps(ctx, sort_keys=True).encode()).hexdigest()
|
|
item, _ = Investigation.objects.update_or_create(
|
|
dataset=ds, user=request.user, key=key, defaults={"context": ctx, "status": status, "note": note}
|
|
)
|
|
audit(request, "investigation.update", ds.pk, {"key": key, "status": status})
|
|
return JsonResponse({"id": item.pk, "key": key, "status": status})
|
|
|
|
|
|
@endpoint(("GET", "POST", "PATCH"), roles=("admin",))
|
|
def users(request):
|
|
if request.member.departments:
|
|
raise APIError("限域管理员不能管理账号", 403)
|
|
if request.method == "GET":
|
|
return JsonResponse(
|
|
{
|
|
"items": [
|
|
{
|
|
"id": m.user_id,
|
|
"username": m.user.username,
|
|
"role": m.role,
|
|
"active": m.user.is_active,
|
|
"departments": m.departments,
|
|
}
|
|
for m in Membership.objects.filter(
|
|
organization=request.member.organization
|
|
).select_related("user")
|
|
]
|
|
}
|
|
)
|
|
data = body(request)
|
|
role, departments = data.get("role"), data.get("departments", [])
|
|
if (
|
|
role not in ("admin", "analyst", "viewer")
|
|
or not isinstance(departments, list)
|
|
or any(not isinstance(d, str) or not d or len(d) > 256 for d in departments)
|
|
or len(departments) > 100
|
|
):
|
|
raise APIError("角色或科室范围无效")
|
|
with transaction.atomic():
|
|
if request.method == "POST":
|
|
username, password = data.get("username"), data.get("password")
|
|
if (
|
|
not isinstance(username, str)
|
|
or not 1 <= len(username) <= 150
|
|
or not isinstance(password, str)
|
|
):
|
|
raise APIError("账号格式无效")
|
|
validate_password(password, User(username=username))
|
|
try:
|
|
user = User.objects.create_user(username=username, password=password)
|
|
except IntegrityError as exc:
|
|
raise APIError("用户名已存在", 409) from exc
|
|
Membership.objects.create(
|
|
user=user, organization=request.member.organization, role=role, departments=departments
|
|
)
|
|
else:
|
|
if type(data.get("id")) is not int or data["id"] <= 0:
|
|
raise APIError("账号 ID 无效")
|
|
membership = (
|
|
Membership.objects.filter(user_id=data.get("id"), organization=request.member.organization)
|
|
.select_related("user")
|
|
.first()
|
|
)
|
|
if not membership:
|
|
raise APIError("账号不存在", 404)
|
|
if membership.user_id == request.user.pk:
|
|
raise APIError("不可在此修改自己的权限")
|
|
if type(data.get("active")) is not bool:
|
|
raise APIError("账号状态无效")
|
|
user = membership.user
|
|
user.is_active = data["active"]
|
|
user.save(update_fields=["is_active"])
|
|
membership.role, membership.departments = role, departments
|
|
membership.save()
|
|
audit(request, "user.update", user.pk, {"role": role, "departments": departments})
|
|
return JsonResponse({"ok": True})
|
|
|
|
|
|
@endpoint(roles=("admin",))
|
|
def audit_log(request):
|
|
if request.member.departments:
|
|
raise APIError("没有组织审计权限", 403)
|
|
return JsonResponse(
|
|
{
|
|
"items": list(
|
|
Audit.objects.filter(organization=request.member.organization)
|
|
.order_by("-id")
|
|
.values("id", "user__username", "action", "resource", "details", "created_at")[:200]
|
|
)
|
|
}
|
|
)
|
|
|
|
|
|
@endpoint(public=True)
|
|
def health(request):
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("SELECT 1")
|
|
return JsonResponse({"status": "ok"})
|
|
|
|
|
|
def index(request):
|
|
path = settings.BASE_DIR / "frontend/dist/index.html"
|
|
if not path.exists():
|
|
return HttpResponse("Frontend not built. Run npm run build.", status=503)
|
|
return FileResponse(path.open("rb"), content_type="text/html")
|