shishengliang
65904c8270
Preserve August hydrogen revision and September audit evidence; calculate interrupted vehicle history across all prior months. Record remaining carry-policy and layout-baseline discrepancies.
127 lines
5.8 KiB
Python
127 lines
5.8 KiB
Python
"""本机 HTTP API；不可直接作为公网生产服务使用。"""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import tempfile
|
|
from pathlib import Path
|
|
from urllib.parse import parse_qs
|
|
from zipfile import BadZipFile
|
|
|
|
from . import VERSION
|
|
from .core import build
|
|
from .workbooks import leases, mileage, export
|
|
|
|
|
|
def encoded(value):
|
|
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
|
|
|
|
|
class Store:
|
|
def __init__(self, directory):
|
|
self.directory = Path(directory)
|
|
self.directory.mkdir(parents=True, exist_ok=True)
|
|
|
|
def save(self, kind, value):
|
|
content = encoded(value)
|
|
identifier = hashlib.sha256(content).hexdigest()
|
|
folder = self.directory / kind
|
|
folder.mkdir(exist_ok=True)
|
|
target = folder / (identifier + ".json")
|
|
# 同一输入重复提交得到同一标识；所有记录以内容寻址，旧快照不被逻辑更新。
|
|
if not target.exists():
|
|
fd, temporary = tempfile.mkstemp(dir=folder)
|
|
try:
|
|
with os.fdopen(fd, "wb") as f:
|
|
f.write(content)
|
|
os.replace(temporary, target)
|
|
finally:
|
|
if os.path.exists(temporary):
|
|
os.unlink(temporary)
|
|
return identifier
|
|
|
|
def load(self, kind, identifier):
|
|
if not re.fullmatch(r"[0-9a-f]{64}", identifier):
|
|
raise ValueError("非法快照标识")
|
|
p = self.directory / kind / (identifier + ".json")
|
|
if not p.is_file():
|
|
raise FileNotFoundError(identifier)
|
|
return dict(json.loads(p.read_text()), id=identifier)
|
|
|
|
|
|
class App:
|
|
def __init__(self, directory):
|
|
self.store = Store(directory)
|
|
|
|
def __call__(self, env, respond):
|
|
code = 200
|
|
mime = "application/json; charset=utf-8"
|
|
try:
|
|
path, method = env.get("PATH_INFO", ""), env["REQUEST_METHOD"]
|
|
query = parse_qs(env.get("QUERY_STRING", ""))
|
|
if method == "GET" and path == "/health":
|
|
payload = dict(status="ok", version=VERSION)
|
|
elif method == "POST" and path in ("/imports/leases", "/imports/mileage"):
|
|
blob = self.body(env)
|
|
if path.endswith("/leases"):
|
|
payload = leases(blob)
|
|
else:
|
|
payload = mileage(blob, query.get("source", [""])[0], query.get("provider", [""])[0])
|
|
payload["sha256"] = hashlib.sha256(blob).hexdigest()
|
|
payload["parser_version"] = VERSION
|
|
identifier = self.store.save("imports", payload)
|
|
payload = dict(id=identifier, kind=payload["kind"], raw_rows=payload["raw_rows"],
|
|
accepted_rows=len(payload["rows"]), issues=payload["issues"])
|
|
code = 201
|
|
elif method == "GET" and path.startswith("/imports/"):
|
|
payload = self.store.load("imports", path.removeprefix("/imports/"))
|
|
elif method == "POST" and path == "/assessments":
|
|
request = json.loads(self.body(env))
|
|
if not isinstance(request, dict) or not isinstance(request.get("config"), dict):
|
|
raise ValueError("请求必须为对象，且包含 config 对象")
|
|
lease = self.store.load("imports", request["lease_import_id"])
|
|
if lease["kind"] != "leases":
|
|
raise ValueError("lease_import_id 类型错误")
|
|
mileages = [self.store.load("imports", i) for i in request["mileage_import_ids"]]
|
|
if any(i["kind"] != "mileage" for i in mileages):
|
|
raise ValueError("mileage_import_ids 类型错误")
|
|
result = build(lease, mileages, request["config"])
|
|
result.update(engine_version=VERSION, inputs=request)
|
|
identifier = self.store.save("assessments", result)
|
|
payload = dict(id=identifier, status=result["status"], summary=result["summary"], issues=result["issues"])
|
|
code = 201
|
|
elif method == "GET" and path.startswith("/assessments/"):
|
|
tail = path.removeprefix("/assessments/")
|
|
wants_excel = tail.endswith("/export.xlsx")
|
|
identifier = tail.removesuffix("/export.xlsx") if wants_excel else tail
|
|
result = self.store.load("assessments", identifier)
|
|
if wants_excel:
|
|
if result["status"] != "ready":
|
|
code, payload = 409, dict(error="仍有待核对问题", issues=result["issues"])
|
|
else:
|
|
payload = export(result)
|
|
mime = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
|
else:
|
|
payload = result
|
|
else:
|
|
code, payload = 404, dict(error="接口不存在")
|
|
except FileNotFoundError:
|
|
code, payload = 404, dict(error="快照不存在")
|
|
except (ValueError, KeyError, TypeError, BadZipFile) as exc:
|
|
code, payload = 400, dict(error=str(exc))
|
|
data = payload if isinstance(payload, bytes) else encoded(payload)
|
|
labels = {200: "OK", 201: "Created", 400: "Bad Request", 404: "Not Found", 409: "Conflict"}
|
|
respond(f"{code} {labels[code]}", [("Content-Type", mime), ("Content-Length", str(len(data))),
|
|
("Cache-Control", "no-store")])
|
|
return [data]
|
|
|
|
@staticmethod
|
|
def body(env):
|
|
size = int(env.get("CONTENT_LENGTH") or 0)
|
|
if not 0 < size <= 20 * 1024 * 1024:
|
|
raise ValueError("请求体需大于 0 且不超过 20 MB")
|
|
content = env["wsgi.input"].read(size)
|
|
if len(content) != size:
|
|
raise ValueError("请求体不完整")
|
|
return content
|