from __future__ import annotations import csv import io import json import sqlite3 from datetime import datetime from http import HTTPStatus from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any, Dict, Tuple from urllib.parse import urlparse from pricing import PRICING_TYPE_CONFIG, PriceInput, build_price_input, calculate_pricing PROJECT_ROOT = Path(__file__).resolve().parent STATIC_DIR = PROJECT_ROOT / "static" DATA_DIR = PROJECT_ROOT / "data" DB_PATH = DATA_DIR / "pricing_records.sqlite3" DB_MIGRATIONS: Dict[str, str] = { "pricing_type": "TEXT NOT NULL DEFAULT 'liangzhu_self_developed'", "pricing_type_label": "TEXT NOT NULL DEFAULT '良渚自研产品'", "ip_rate": "REAL NOT NULL DEFAULT 0", "target_profit_rate": "REAL NOT NULL DEFAULT 0", "ip_fee": "REAL NOT NULL DEFAULT 0", "target_profit_amount": "REAL NOT NULL DEFAULT 0", } def init_db() -> None: DATA_DIR.mkdir(parents=True, exist_ok=True) with sqlite3.connect(DB_PATH) as conn: conn.execute( """ CREATE TABLE IF NOT EXISTS pricing_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at TEXT NOT NULL, pricing_type TEXT NOT NULL DEFAULT 'liangzhu_self_developed', pricing_type_label TEXT NOT NULL DEFAULT '良渚自研产品', product_name TEXT NOT NULL, tax_included_cost REAL NOT NULL, freight REAL NOT NULL, discount_rate REAL NOT NULL, ip_rate REAL NOT NULL DEFAULT 0, platform_rate REAL NOT NULL, promotion_rate REAL NOT NULL, team_rate REAL NOT NULL, vat_rate REAL NOT NULL, target_profit_rate REAL NOT NULL DEFAULT 0, break_even_price_raw REAL NOT NULL, break_even_price_ceil INTEGER NOT NULL, list_price_raw REAL NOT NULL, list_price_ceil INTEGER NOT NULL, discount_sale_price REAL NOT NULL, discount_break_even INTEGER NOT NULL, ip_fee REAL NOT NULL DEFAULT 0, target_profit_amount REAL NOT NULL DEFAULT 0, final_list_price REAL, note TEXT, input_json TEXT NOT NULL, result_json TEXT NOT NULL ) """ ) existing_columns = { row[1] for row in conn.execute("PRAGMA table_info(pricing_records)").fetchall() } for column, ddl in DB_MIGRATIONS.items(): if column not in existing_columns: conn.execute(f"ALTER TABLE pricing_records ADD COLUMN {column} {ddl}") conn.commit() def pricing_types_payload() -> Dict[str, Any]: return { "pricing_types": [ {"value": key, **config} for key, config in PRICING_TYPE_CONFIG.items() ] } def record_to_dict(row: sqlite3.Row) -> Dict[str, Any]: data = dict(row) data["discount_break_even"] = bool(data["discount_break_even"]) if not data.get("pricing_type"): data["pricing_type"] = "liangzhu_self_developed" if not data.get("pricing_type_label"): data["pricing_type_label"] = "良渚自研产品" return data def save_record(item: PriceInput, result: Dict[str, Any]) -> int: created_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with sqlite3.connect(DB_PATH) as conn: cur = conn.execute( """ INSERT INTO pricing_records ( created_at, pricing_type, pricing_type_label, product_name, tax_included_cost, freight, discount_rate, ip_rate, platform_rate, promotion_rate, team_rate, vat_rate, target_profit_rate, break_even_price_raw, break_even_price_ceil, list_price_raw, list_price_ceil, discount_sale_price, discount_break_even, ip_fee, target_profit_amount, final_list_price, note, input_json, result_json ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( created_at, item.pricing_type, result["pricing_type_label"], item.product_name, item.tax_included_cost, item.freight, item.discount_rate, item.ip_rate, item.platform_rate, item.promotion_rate, item.team_rate, item.vat_rate, item.target_profit_rate, result["break_even_price_raw"], result["break_even_price_ceil"], result["list_price_raw"], result["list_price_ceil"], result["discount_sale_price"], 1 if result["discount_break_even"] else 0, result["ip_fee"], result["target_profit_amount"], item.final_list_price, item.note, json.dumps(item.__dict__, ensure_ascii=False), json.dumps(result, ensure_ascii=False), ), ) conn.commit() return int(cur.lastrowid) def list_records() -> Dict[str, Any]: with sqlite3.connect(DB_PATH) as conn: conn.row_factory = sqlite3.Row rows = conn.execute( """ SELECT id, created_at, pricing_type, pricing_type_label, product_name, tax_included_cost, freight, discount_rate, ip_rate, platform_rate, promotion_rate, team_rate, vat_rate, target_profit_rate, break_even_price_raw, break_even_price_ceil, list_price_raw, list_price_ceil, discount_sale_price, discount_break_even, ip_fee, target_profit_amount, final_list_price, note FROM pricing_records ORDER BY id DESC LIMIT 200 """ ).fetchall() return {"records": [record_to_dict(row) for row in rows], "db_path": str(DB_PATH)} def export_csv_text() -> str: records = list_records()["records"] output = io.StringIO() fieldnames = [ "id", "created_at", "pricing_type_label", "product_name", "tax_included_cost", "freight", "discount_rate", "ip_rate", "platform_rate", "promotion_rate", "team_rate", "vat_rate", "target_profit_rate", "break_even_price_raw", "break_even_price_ceil", "list_price_raw", "list_price_ceil", "discount_sale_price", "discount_break_even", "ip_fee", "target_profit_amount", "final_list_price", "note", ] writer = csv.DictWriter(output, fieldnames=fieldnames) writer.writeheader() writer.writerows(records) return output.getvalue() class PricingHandler(SimpleHTTPRequestHandler): def __init__(self, *args: Any, **kwargs: Any) -> None: super().__init__(*args, directory=str(STATIC_DIR), **kwargs) def log_message(self, fmt: str, *args: Any) -> None: print("[%s] %s" % (datetime.now().strftime("%H:%M:%S"), fmt % args)) def _send_json(self, payload: Dict[str, Any], status: int = 200) -> None: body = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _send_text(self, text: str, content_type: str, status: int = 200) -> None: body = text.encode("utf-8-sig" if content_type.startswith("text/csv") else "utf-8") self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _read_json(self) -> Tuple[Dict[str, Any], str | None]: length = int(self.headers.get("Content-Length", "0") or 0) raw = self.rfile.read(length).decode("utf-8") if length else "{}" try: payload = json.loads(raw) except json.JSONDecodeError as exc: return {}, f"JSON 格式错误：{exc}" if not isinstance(payload, dict): return {}, "请求体必须是 JSON object" return payload, None def do_GET(self) -> None: path = urlparse(self.path).path if path == "/api/health": self._send_json({"ok": True, "db_path": str(DB_PATH)}) elif path == "/api/pricing-types": self._send_json(pricing_types_payload()) elif path == "/api/records": self._send_json(list_records()) elif path == "/api/records.csv": self._send_text(export_csv_text(), "text/csv; charset=utf-8") else: super().do_GET() def do_POST(self) -> None: path = urlparse(self.path).path payload, error = self._read_json() if error: self._send_json({"ok": False, "error": error}, HTTPStatus.BAD_REQUEST) return try: item = build_price_input(payload) result = calculate_pricing(item).to_dict() except ValueError as exc: self._send_json({"ok": False, "error": str(exc)}, HTTPStatus.BAD_REQUEST) return if path == "/api/calculate": self._send_json({"ok": True, "result": result}) elif path == "/api/records": record_id = save_record(item, result) self._send_json({"ok": True, "id": record_id, "result": result, "db_path": str(DB_PATH)}) else: self._send_json({"ok": False, "error": "接口不存在"}, HTTPStatus.NOT_FOUND) def main() -> None: import argparse parser = argparse.ArgumentParser(description="良渚产品电商定价计算器") parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8766) args = parser.parse_args() init_db() server = ThreadingHTTPServer((args.host, args.port), PricingHandler) print(f"良渚产品电商定价计算器：http://{args.host}:{args.port}/") print(f"核算记录数据库：{DB_PATH}") try: server.serve_forever() except KeyboardInterrupt: print("\nserver stopped") finally: server.server_close() if __name__ == "__main__": main()