From 6327ef7641051b55d398b0e9aa6fe116dd19b4d7 Mon Sep 17 00:00:00 2001 From: star Date: Thu, 10 Sep 2026 16:18:15 +0800 Subject: [PATCH] =?UTF-8?q?init:=20=E5=8D=9A=E7=89=B9=E7=BB=B4=E9=87=87?= =?UTF-8?q?=E8=B4=AD=E8=AE=A2=E5=8D=95=E7=B3=BB=E7=BB=9F=20v1.0=20?= =?UTF-8?q?=E6=BA=90=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 38 + po_app/amount_cn.py | 89 ++ po_app/app.py | 1071 +++++++++++++++++++++++ po_app/models.py | 154 ++++ po_app/pdf_export.py | 397 +++++++++ po_app/requirements.txt | 5 + po_app/static/style.css | 121 +++ po_app/templates/base.html | 36 + po_app/templates/index.html | 50 ++ po_app/templates/login.html | 31 + po_app/templates/materials.html | 125 +++ po_app/templates/order_edit.html | 201 +++++ po_app/templates/order_new.html | 30 + po_app/templates/order_view.html | 203 +++++ po_app/templates/orders.html | 53 ++ po_app/templates/stats.html | 40 + po_app/templates/suppliers.html | 132 +++ po_app/templates/users.html | 39 + po_app/使用说明.md | 77 ++ po_app/博特维采购订单系统.spec | 51 ++ po_app/博特维采购订单系统_单文件版.spec | 45 + po_app/启动.bat | 16 + tools/dump_xls.py | 40 + tools/e2e_test.py | 99 +++ tools/git_init_commit.py | 49 ++ tools/pdf_preview.py | 28 + tools/run_check.py | 13 + tools/test_approve_stamp.py | 100 +++ tools/test_batch_import.py | 208 +++++ tools/test_edit_row.py | 119 +++ tools/test_import.py | 73 ++ tools/test_pages.py | 28 + tools/test_stamp_toggle.py | 156 ++++ tools/verify_exe.py | 82 ++ tools/xls_layout.py | 30 + 35 files changed, 4029 insertions(+) create mode 100644 .gitignore create mode 100644 po_app/amount_cn.py create mode 100644 po_app/app.py create mode 100644 po_app/models.py create mode 100644 po_app/pdf_export.py create mode 100644 po_app/requirements.txt create mode 100644 po_app/static/style.css create mode 100644 po_app/templates/base.html create mode 100644 po_app/templates/index.html create mode 100644 po_app/templates/login.html create mode 100644 po_app/templates/materials.html create mode 100644 po_app/templates/order_edit.html create mode 100644 po_app/templates/order_new.html create mode 100644 po_app/templates/order_view.html create mode 100644 po_app/templates/orders.html create mode 100644 po_app/templates/stats.html create mode 100644 po_app/templates/suppliers.html create mode 100644 po_app/templates/users.html create mode 100644 po_app/使用说明.md create mode 100644 po_app/博特维采购订单系统.spec create mode 100644 po_app/博特维采购订单系统_单文件版.spec create mode 100755 po_app/启动.bat create mode 100644 tools/dump_xls.py create mode 100644 tools/e2e_test.py create mode 100644 tools/git_init_commit.py create mode 100644 tools/pdf_preview.py create mode 100644 tools/run_check.py create mode 100644 tools/test_approve_stamp.py create mode 100644 tools/test_batch_import.py create mode 100644 tools/test_edit_row.py create mode 100644 tools/test_import.py create mode 100644 tools/test_pages.py create mode 100644 tools/test_stamp_toggle.py create mode 100644 tools/verify_exe.py create mode 100644 tools/xls_layout.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..2f1b78a --- /dev/null +++ b/.gitignore @@ -0,0 +1,38 @@ +# ===== Python ===== +__pycache__/ +*.py[cod] +*.pyo +*.egg-info/ +.eggs/ +build/ +venv/ +.venv/ +env/ + +# ===== PyInstaller 构建产物（可重新生成，不入库）===== +dist/ +dist_backup_*/ +*.manifest +*.pyz + +# ===== 运行时数据：含真实业务数据，绝不入库 ===== +instance/ +*.db +*.sqlite +*.sqlite3 +uploads/ +exports/ + +# ===== 测试产生的样张 ===== +tools/out/ + +# ===== 编辑器 / 系统 ===== +.vscode/ +.idea/ +*.swp +.DS_Store +Thumbs.db +*.log + +# ===== 本地工具目录 ===== +.workbuddy/ diff --git a/po_app/amount_cn.py b/po_app/amount_cn.py new file mode 100644 index 0000000..e08243c --- /dev/null +++ b/po_app/amount_cn.py @@ -0,0 +1,89 @@ +# -*- coding: utf-8 -*- +"""人民币金额转中文大写。""" + + +def amount_to_cn(amount): + """1234.56 -> 壹仟贰佰叁拾肆元伍角陆分""" + digits = u"零壹贰叁肆伍陆柒捌玖" + units = [u"", u"拾", u"佰", u"仟"] + big_units = [u"", u"万", u"亿", u"兆"] + + try: + amount = round(float(amount) + 0.0, 2) + except (TypeError, ValueError): + return u"零元整" + + negative = amount < 0 + amount = abs(amount) + integer_part = int(amount) + # 避免浮点误差：分位用字符串处理 + fen = int(round((amount - integer_part) * 100)) + if fen == 100: + integer_part += 1 + fen = 0 + + def int_section(num): + """0-9999 转中文""" + s = u"" + zero_flag = False + str_num = "%04d" % num + for i, ch in enumerate(str_num): + d = int(ch) + pos = 3 - i + if d == 0: + zero_flag = True + else: + if zero_flag and s: + s += digits[0] + zero_flag = False + s += digits[d] + units[pos] + return s + + if integer_part == 0: + result = u"" + else: + sections = [] + n = integer_part + while n > 0: + sections.append(n % 10000) + n //= 10000 + parts = [] + for idx, sec in enumerate(sections): + if sec == 0: + parts.append(u"") + continue + parts.append(int_section(sec) + big_units[idx]) + result = u"" + for idx in range(len(parts) - 1, -1, -1): + p = parts[idx] + if not p: + if result and not result.endswith(digits[0]): + result += digits[0] + continue + if result and idx < len(parts) - 1 and sections[idx] < 1000: + result += digits[0] + result += p + result = result.rstrip(digits[0]) or u"零" + + if fen == 0: + result += u"元整" if result else u"零元整" + else: + jiao = fen // 10 + f = fen % 10 + if result: + result += u"元" + if jiao: + result += digits[jiao] + u"角" + elif f and result: + result += digits[0] + if f: + result += digits[f] + u"分" + if negative: + result = u"负" + result + return result + + +if __name__ == "__main__": + for v in (0, 0.05, 0.5, 1, 10, 100.5, 1234.56, 10000, 1002003, 20340.07, + 100000000, 10500.5): + print(v, amount_to_cn(v)) diff --git a/po_app/app.py b/po_app/app.py new file mode 100644 index 0000000..3677bd2 --- /dev/null +++ b/po_app/app.py @@ -0,0 +1,1071 @@ +# -*- coding: utf-8 -*- +"""博特维采购订单管理系统：制单 → 主管审核 → 经理批准 → 导出 PDF。""" +import io +import os +import sys +from datetime import datetime +from functools import wraps + +from flask import (Flask, flash, jsonify, redirect, render_template, request, + send_file, send_from_directory, session, url_for) +from sqlalchemy import func +from werkzeug.security import check_password_hash, generate_password_hash +from werkzeug.utils import secure_filename + +from amount_cn import amount_to_cn +from models import (COMPANY_NAME, DOC_TYPES, ROLE_TEXT, STATUS_TEXT, + DEFAULT_DELIVERY_PLACE, DEFAULT_OUR_CONTACT, + DEFAULT_OUR_PHONE, DEFAULT_OUR_TAX_NO, DEFAULT_TAX_RATE, + ApprovalLog, Material, Order, OrderItem, Supplier, User, + init_db) +from pdf_export import build_pdf + +# 打包成 EXE 后：资源在临时目录 _MEIPASS，数据必须写在 EXE 同级目录 +if getattr(sys, "frozen", False): + RES_DIR = sys._MEIPASS + BASE_DIR = os.path.dirname(sys.executable) +else: + RES_DIR = os.path.dirname(os.path.abspath(__file__)) + BASE_DIR = RES_DIR + +INSTANCE = os.path.join(BASE_DIR, "instance") +EXPORTS = os.path.join(BASE_DIR, "exports") +UPLOADS = os.path.join(BASE_DIR, "uploads") +for d in (INSTANCE, EXPORTS, UPLOADS): + os.makedirs(d, exist_ok=True) + +app = Flask(__name__, + template_folder=os.path.join(RES_DIR, "templates"), + static_folder=os.path.join(RES_DIR, "static")) +app.secret_key = "botw-po-system-2026" +app.config["MAX_CONTENT_LENGTH"] = 8 * 1024 * 1024 + +engine, Session = init_db(os.path.join(INSTANCE, "po.db").replace("\\", "/")) +db = Session() + + +def now(): + return datetime.now() + + +# ---------------- 基础工具 ---------------- +def current_user(): + uid = session.get("user_id") + if not uid: + return None + return db.get(User, uid) + + +def login_required(f): + @wraps(f) + def wrapper(*a, **kw): + if not current_user(): + return redirect(url_for("login")) + return f(*a, **kw) + return wrapper + + +def role_required(*roles): + def deco(f): + @wraps(f) + def wrapper(*a, **kw): + u = current_user() + if not u: + return redirect(url_for("login")) + if u.role != "admin" and u.role not in roles: + flash(u"当前账号无此操作权限", "error") + return redirect(url_for("index")) + return f(*a, **kw) + return wrapper + return deco + + +def add_log(order, action, user, comment=None): + db.add(ApprovalLog(order_id=order.id, action=action, + user_name=user.name if user else u"系统", + comment=comment, created_at=now())) + + +def can_edit(order, user): + if not user: + return False + if user.role == "admin": + return order.status in ("draft", "rejected") + return (order.status in ("draft", "rejected") + and order.created_by == user.id) + + +def next_order_no(doc_type): + """单号规则：PO/WO + 年月日 + 4 位递增数字。 + + 序号解析兼容旧格式（PO20260909-001 与 PO202609090001 都能正确读出）， + 这样改规则后历史单据不会导致序号从头开始。 + """ + prefix = "PO" if doc_type == "purchase" else "WO" + head = "%s%s" % (prefix, now().strftime("%Y%m%d")) + seq = 0 + for o in db.query(Order).filter(Order.order_no.like("%s%%" % head)).all(): + tail = (o.order_no[len(head):] or "").lstrip("-") + try: + n = int(tail) + except ValueError: + continue + if n > seq: + seq = n + return "%s%04d" % (head, seq + 1) + + +def recalc(order): + total = 0.0 + for it in order.items: + try: + qty = float(it.qty or 0) + price = float(it.price or 0) + except (TypeError, ValueError): + qty = price = 0.0 + it.amount = round(qty * price, 2) + total += it.amount + order.total_amount = round(total, 2) + return total + + +def parse_items_from_form(form, order): + codes = form.getlist("item_code") + specs = form.getlist("item_spec") + units = form.getlist("item_unit") + qtys = form.getlist("item_qty") + prices = form.getlist("item_price") + dates = form.getlist("item_date") + remarks = form.getlist("item_remark") + order.items = [] + for i in range(len(codes)): + code = (codes[i] or "").strip() + spec = (specs[i] or "").strip() + if not code and not spec: + continue + def f(v): + try: + return float(v) + except (TypeError, ValueError): + return 0.0 + qty = f(qtys[i] if i < len(qtys) else 0) + price = f(prices[i] if i < len(prices) else 0) + order.items.append(OrderItem( + seq=i + 1, material_code=code, material_spec=spec, + unit=(units[i] if i < len(units) else "").strip(), + qty=qty, price=price, amount=round(qty * price, 2), + delivery_date=(dates[i] if i < len(dates) else "").strip(), + remark=(remarks[i] if i < len(remarks) else "").strip())) + recalc(order) + + +# ---------------- 初始化 ---------------- +def seed(): + if db.query(User).count(): + return + users = [ + ("admin", "admin123", u"系统管理员", "admin"), + ("maker", "123456", u"张采购", "maker"), + ("reviewer", "123456", u"李主管", "reviewer"), + ("approver", "123456", u"王经理", "approver"), + ] + for un, pw, name, role in users: + db.add(User(username=un, password_hash=generate_password_hash(pw), + name=name, role=role)) + db.add(Supplier(code="S001", name=u"苏州精锐五金制品有限公司", contact=u"周经理", + phone="13800000001", tax_no="91320500MA1XXXXXXX", + bank_name=u"中国银行苏州工业园区支行", + bank_account="4562 1088 0012 3456", payment_terms=u"月结30天")) + db.add(Supplier(code="S002", name=u"昆山宏远电子科技有限公司", contact=u"陈主管", + phone="13900000002", tax_no="91320583MA2XXXXXXX", + bank_name=u"建设银行昆山分行", + bank_account="3220 1988 2233 4455", payment_terms=u"货到付款")) + db.add(Material(code="M-1001", spec=u"铝合金支架 6061-T6 200×80×10mm", unit=u"件", price=38.5)) + db.add(Material(code="M-1002", spec=u"步进电机 57BYG250 2相", unit=u"台", price=265.0)) + db.add(Material(code="M-1003", spec=u"同步带 5M-500 宽15mm", unit=u"条", price=46.0)) + db.commit() + + +seed() + + +# ---------------- 认证 ---------------- +@app.route("/login", methods=["GET", "POST"]) +def login(): + if request.method == "POST": + u = db.query(User).filter_by(username=request.form.get("username", "").strip()).first() + if u and u.active and check_password_hash(u.password_hash, request.form.get("password", "")): + session["user_id"] = u.id + session["user_name"] = u.name + session["role"] = u.role + return redirect(url_for("index")) + flash(u"用户名或密码错误", "error") + return render_template("login.html") + + +@app.route("/logout") +def logout(): + session.clear() + return redirect(url_for("login")) + + +# ---------------- 首页 ---------------- +@app.route("/") +@login_required +def index(): + u = current_user() + todo = [] + if u.role in ("reviewer", "admin"): + todo = db.query(Order).filter_by(status="pending_review").order_by(Order.id.desc()).all() + if u.role in ("approver", "admin"): + todo += db.query(Order).filter_by(status="pending_approval").order_by(Order.id.desc()).all() + my_orders = [] + if u.role in ("maker", "admin"): + my_orders = (db.query(Order).filter_by(created_by=u.id) + .order_by(Order.id.desc()).limit(8).all()) + counts = {s: db.query(Order).filter_by(status=s).count() for s in STATUS_TEXT} + month_total = (db.query(func.sum(Order.total_amount)) + .filter(Order.status == "approved").scalar() or 0) + return render_template("index.html", user=u, todo=todo, my_orders=my_orders, + counts=counts, month_total=month_total, + status_text=STATUS_TEXT, role_text=ROLE_TEXT) + + +# ---------------- 订单 ---------------- +@app.route("/orders") +@login_required +def orders(): + u = current_user() + q = db.query(Order) + kw = request.args.get("kw", "").strip() + status = request.args.get("status", "") + doc_type = request.args.get("doc_type", "") + date_from = request.args.get("date_from", "") + date_to = request.args.get("date_to", "") + if kw: + q = q.filter(Order.order_no.like("%%%s%%" % kw) | + Order.supplier_name.like("%%%s%%" % kw)) + if status: + q = q.filter(Order.status == status) + if doc_type: + q = q.filter(Order.doc_type == doc_type) + if date_from: + q = q.filter(Order.order_date >= date_from) + if date_to: + q = q.filter(Order.order_date <= date_to) + if u.role == "maker": + q = q.filter(Order.created_by == u.id) + rows = q.order_by(Order.id.desc()).all() + return render_template("orders.html", user=u, rows=rows, status_text=STATUS_TEXT, + doc_types=DOC_TYPES, args=request.args) + + +@app.route("/orders/new", methods=["GET", "POST"]) +@login_required +def order_new(): + u = current_user() + suppliers = db.query(Supplier).filter_by(active=True).all() + if request.method == "POST": + doc_type = request.form.get("doc_type", "purchase") + o = Order(order_no=next_order_no(doc_type), doc_type=doc_type, + status="draft", created_by=u.id, created_by_name=u.name, + created_at=now(), order_date=now().strftime("%Y-%m-%d"), + our_company=COMPANY_NAME, delivery_place=DEFAULT_DELIVERY_PLACE, + our_contact=DEFAULT_OUR_CONTACT, our_phone=DEFAULT_OUR_PHONE, + our_tax_no=DEFAULT_OUR_TAX_NO, tax_rate=DEFAULT_TAX_RATE) + sid = request.form.get("supplier_id") + if sid: + s = db.get(Supplier, int(sid)) + if s: + o.supplier_name = s.name + o.supplier_contact = s.contact + o.supplier_phone = s.phone + o.supplier_tax_no = s.tax_no + o.supplier_bank_name = s.bank_name + o.supplier_bank_account = s.bank_account + o.payment_terms = s.payment_terms + db.add(o) + db.commit() + return redirect(url_for("order_edit", oid=o.id)) + return render_template("order_new.html", user=u, suppliers=suppliers, + doc_types=DOC_TYPES) + + +@app.route("/orders/", methods=["GET", "POST"]) +@app.route("/orders//edit", methods=["GET", "POST"]) +@login_required +def order_edit(oid): + u = current_user() + o = db.get(Order, oid) + if not o: + flash(u"订单不存在", "error") + return redirect(url_for("orders")) + editable = can_edit(o, u) + if request.method == "POST": + if not editable: + flash(u"当前状态不可编辑", "error") + return redirect(url_for("order_view", oid=o.id)) + for f in ("supplier_name", "supplier_contact", "supplier_phone", + "supplier_tax_no", "supplier_bank_name", "supplier_bank_account", + "payment_terms", "our_company", "our_contact", "our_phone", + "our_tax_no", "tax_rate", "order_date", "delivery_place", "note"): + setattr(o, f, request.form.get(f, "").strip()) + parse_items_from_form(request.form, o) + db.commit() + flash(u"已保存", "ok") + return redirect(url_for("order_edit", oid=o.id)) + suppliers = db.query(Supplier).filter_by(active=True).all() + materials = db.query(Material).filter_by(active=True).all() + return render_template("order_edit.html", user=u, o=o, editable=editable, + suppliers=suppliers, materials=materials, + doc_types=DOC_TYPES, total_cn=amount_to_cn(o.total_amount)) + + +@app.route("/orders//view") +@login_required +def order_view(oid): + u = current_user() + o = db.get(Order, oid) + if not o: + flash(u"订单不存在", "error") + return redirect(url_for("orders")) + return render_template("order_view.html", user=u, o=o, editable=can_edit(o, u), + total_cn=amount_to_cn(o.total_amount), + status_text=STATUS_TEXT, role_text=ROLE_TEXT) + + +@app.route("/orders//submit", methods=["POST"]) +@login_required +def order_submit(oid): + u = current_user() + o = db.get(Order, oid) + if not o or not can_edit(o, u): + flash(u"无权限或状态不允许", "error") + return redirect(url_for("orders")) + if not o.items: + flash(u"明细为空，不能提交", "error") + return redirect(url_for("order_edit", oid=o.id)) + o.status = "pending_review" + o.submitted_at = now() + o.reject_reason = None + add_log(o, u"提交审核", u, request.form.get("comment")) + db.commit() + flash(u"已提交，等待主管审核", "ok") + return redirect(url_for("order_view", oid=o.id)) + + +@app.route("/orders//review", methods=["POST"]) +@login_required +@role_required("reviewer") +def order_review(oid): + u = current_user() + o = db.get(Order, oid) + if not o or o.status != "pending_review": + flash(u"状态不允许", "error") + return redirect(url_for("orders")) + # 确认前必须先加盖审核章（服务端校验，防止绕过前端直接提交） + if request.form.get("stamped") != "1": + flash(u"请先加盖审核章，再确认审核", "error") + return redirect(url_for("order_view", oid=o.id)) + o.status = "pending_approval" + o.reviewed_by = u.name + o.reviewed_at = now() + add_log(o, u"主管审核通过（已加盖审核章）", u, request.form.get("comment")) + db.commit() + flash(u"审核通过，等待经理批准", "ok") + return redirect(url_for("order_view", oid=o.id)) + + +@app.route("/orders//approve", methods=["POST"]) +@login_required +@role_required("approver") +def order_approve(oid): + u = current_user() + o = db.get(Order, oid) + if not o or o.status != "pending_approval": + flash(u"状态不允许", "error") + return redirect(url_for("orders")) + # 与主管审核一致：确认前必须先加盖批准章（服务端校验，防绕过） + if request.form.get("stamped") != "1": + flash(u"请先加盖批准章，再确认批准", "error") + return redirect(url_for("order_view", oid=o.id)) + o.status = "approved" + o.approved_by = u.name + o.approved_at = now() + add_log(o, u"经理批准（已加盖批准章）", u, request.form.get("comment")) + db.commit() + flash(u"已批准，可以导出 PDF", "ok") + return redirect(url_for("order_view", oid=o.id)) + + +@app.route("/orders//reject", methods=["POST"]) +@login_required +def order_reject(oid): + u = current_user() + o = db.get(Order, oid) + if not o: + flash(u"订单不存在", "error") + return redirect(url_for("orders")) + if u.role not in ("admin", "reviewer", "approver"): + flash(u"无权限", "error") + return redirect(url_for("order_view", oid=o.id)) + if o.status not in ("pending_review", "pending_approval"): + flash(u"状态不允许驳回", "error") + return redirect(url_for("order_view", oid=o.id)) + reason = request.form.get("comment", "").strip() + if not reason: + flash(u"请填写驳回原因", "error") + return redirect(url_for("order_view", oid=o.id)) + from_status = o.status + o.status = "rejected" + o.reject_reason = reason + o.reviewed_by = None + o.reviewed_at = None + o.approved_by = None + o.approved_at = None + add_log(o, u"驳回（%s）" % (u"审核环节" if from_status == "pending_review" else u"批准环节"), + u, reason) + db.commit() + flash(u"已驳回给制单人", "ok") + return redirect(url_for("order_view", oid=o.id)) + + +@app.route("/orders//copy", methods=["POST"]) +@login_required +def order_copy(oid): + u = current_user() + src = db.get(Order, oid) + if not src: + flash(u"订单不存在", "error") + return redirect(url_for("orders")) + o = Order(order_no=next_order_no(src.doc_type), doc_type=src.doc_type, + status="draft", created_by=u.id, created_by_name=u.name, + created_at=now(), order_date=now().strftime("%Y-%m-%d")) + for f in ("supplier_name", "supplier_contact", "supplier_phone", "supplier_tax_no", + "supplier_bank_name", "supplier_bank_account", "payment_terms", + "our_company", "our_contact", "our_phone", "our_tax_no", "tax_rate", + "delivery_place", "note"): + setattr(o, f, getattr(src, f)) + # 老单据可能没存需方信息，复制时补默认值，避免新单缺项 + if not o.our_contact: + o.our_contact = DEFAULT_OUR_CONTACT + if not o.our_phone: + o.our_phone = DEFAULT_OUR_PHONE + if not o.our_tax_no: + o.our_tax_no = DEFAULT_OUR_TAX_NO + if not o.tax_rate: + o.tax_rate = DEFAULT_TAX_RATE + db.add(o) + db.flush() + for it in src.items: + db.add(OrderItem(order_id=o.id, seq=it.seq, material_code=it.material_code, + material_spec=it.material_spec, unit=it.unit, qty=it.qty, + price=it.price, amount=it.amount, + delivery_date=it.delivery_date, remark=it.remark)) + recalc(o) + db.commit() + flash(u"已复制为新订单 %s" % o.order_no, "ok") + return redirect(url_for("order_edit", oid=o.id)) + + +@app.route("/orders//pdf") +@login_required +def order_pdf(oid): + o = db.get(Order, oid) + if not o: + flash(u"订单不存在", "error") + return redirect(url_for("orders")) + fname = "%s_%s.pdf" % (o.order_no, now().strftime("%H%M%S")) + path = os.path.join(EXPORTS, fname) + build_pdf(o, o.items, path) + dl = request.args.get("dl") == "1" + return send_file(path, as_attachment=dl, + download_name="%s %s.pdf" % (o.doc_type_text, o.order_no)) + + +@app.route("/orders//import_excel", methods=["POST"]) +@login_required +def import_excel(oid): + u = current_user() + o = db.get(Order, oid) + if not o or not can_edit(o, u): + flash(u"无权限或状态不允许", "error") + return redirect(url_for("orders")) + file = request.files.get("file") + if not file or not file.filename: + flash(u"请选择文件", "error") + return redirect(url_for("order_edit", oid=o.id)) + fname = secure_filename(file.filename) + path = os.path.join(UPLOADS, "%s_%s" % (now().strftime("%H%M%S"), fname)) + file.save(path) + try: + rows = read_rows(path) + except Exception as e: + flash(u"读取失败：%s" % e, "error") + return redirect(url_for("order_edit", oid=o.id)) + mode = request.form.get("mode", "append") + if mode == "replace": + o.items = [] + seq = len(o.items) + added = 0 + for r in rows: + if not any(r): + continue + code, spec, unit, qty, price, date, remark = (list(r) + [""] * 7)[:7] + def f(v): + try: + return float(str(v).strip()) + except (TypeError, ValueError): + return 0.0 + seq += 1 + q = f(qty) + p = f(price) + o.items.append(OrderItem(seq=seq, material_code=str(code or "").strip(), + material_spec=str(spec or "").strip(), + unit=str(unit or "").strip(), qty=q, price=p, + amount=round(q * p, 2), + delivery_date=str(date or "").strip(), + remark=str(remark or "").strip())) + added += 1 + recalc(o) + db.commit() + flash(u"已导入 %d 行明细" % added, "ok") + return redirect(url_for("order_edit", oid=o.id)) + + +def read_rows(path): + """读取 xlsx/xls 明细，返回 [代码,规格,单位,数量,单价,交期,备注]。""" + ext = os.path.splitext(path)[1].lower() + grid = [] + if ext in (".xlsx", ".xlsm"): + import openpyxl + wb = openpyxl.load_workbook(path, data_only=True) + for row in wb[wb.sheetnames[0]].iter_rows(values_only=True): + grid.append(["" if c is None else c for c in row]) + else: + import xlrd + bk = xlrd.open_workbook(path) + sh = bk.sheet_by_index(0) + for r in range(sh.nrows): + grid.append([sh.cell_value(r, c) for c in range(sh.ncols)]) + keys = {"code": [u"代码", u"料号", u"物料编码", u"品号"], + "spec": [u"规格", u"名称", u"品名", u"描述"], + "unit": [u"单位"], + "qty": [u"数量", u"数量(QTY)"], + "price": [u"单价", u"含税单价", u"价格"], + "date": [u"交期", u"交货日期", u"日期"], + "remark": [u"备注"]} + order = ["code", "spec", "unit", "qty", "price", "date", "remark"] + header_idx = None + for i, row in enumerate(grid[:8]): + text = " ".join(str(c) for c in row) + if any(k in text for k in keys["code"]) and any(k in text for k in keys["qty"]): + header_idx = i + break + if header_idx is None: + return [[c for c in row[:7]] for row in grid] + heads = [str(c or "").strip() for c in grid[header_idx]] + mapping = {} + for idx, h in enumerate(heads): + for k, ks in keys.items(): + if any(x in h for x in ks): + mapping[k] = idx + break + out = [] + for row in grid[header_idx + 1:]: + vals = [] + for k in order: + i = mapping.get(k) + vals.append(row[i] if (i is not None and i < len(row)) else "") + out.append(vals) + return out + + +def read_table(path, keys, order): + """通用表格读取：按关键词映射表头，返回按 order 排列的二维列表。 + + 与 read_rows() 的区别是表头判定更宽松（命中任意关键词即可）， + 因为物料表 / 供应商表都没有"数量"列，套用 read_rows 会退化成按位置读。 + """ + ext = os.path.splitext(path)[1].lower() + grid = [] + if ext in (".xlsx", ".xlsm"): + import openpyxl + wb = openpyxl.load_workbook(path, data_only=True) + for row in wb[wb.sheetnames[0]].iter_rows(values_only=True): + grid.append(["" if c is None else c for c in row]) + else: + import xlrd + bk = xlrd.open_workbook(path) + sh = bk.sheet_by_index(0) + for r in range(sh.nrows): + grid.append([sh.cell_value(r, c) for c in range(sh.ncols)]) + header_idx = None + for i, row in enumerate(grid[:8]): + text = " ".join(str(c) for c in row) + if any(any(k in text for k in ks) for ks in keys.values()): + header_idx = i + break + if header_idx is None: + return [list(row[:len(order)]) for row in grid] + heads = [str(c or "").strip() for c in grid[header_idx]] + mapping = {} + for idx, h in enumerate(heads): + for k, ks in keys.items(): + if k in mapping: + continue + if any(x in h for x in ks): + mapping[k] = idx + break + out = [] + for row in grid[header_idx + 1:]: + vals = [] + for k in order: + i = mapping.get(k) + vals.append(row[i] if (i is not None and i < len(row)) else "") + out.append(vals) + return out + + +MATERIAL_KEYS = { + "code": [u"物料代码", u"料号", u"物料编码", u"品号", u"代码"], + "spec": [u"规格", u"名称", u"品名", u"描述"], + "unit": [u"单位"], + "price": [u"单价", u"价格"], +} +MATERIAL_ORDER = ["code", "spec", "unit", "price"] + +SUPPLIER_KEYS = { + "code": [u"编号", u"供应商编号", u"代码"], + "name": [u"名称", u"供应商名称", u"公司"], + "contact": [u"联系人"], + "phone": [u"电话", u"手机", u"联系方式"], + "payment_terms": [u"付款方式", u"结算方式", u"账期"], + "tax_no": [u"税号", u"纳税人识别号"], + "bank_name": [u"开户银行", u"银行名称"], + "bank_account": [u"银行帐号", u"银行账号", u"帐号", u"账号"], +} +SUPPLIER_ORDER = ["code", "name", "contact", "phone", "payment_terms", + "tax_no", "bank_name", "bank_account"] + + +def read_material_rows(path): + """读取物料表，返回 [代码, 规格, 单位, 单价]。""" + return read_table(path, MATERIAL_KEYS, MATERIAL_ORDER) + + +def read_supplier_rows(path): + """读取供应商表，返回 [编号,名称,联系人,电话,付款方式,税号,开户银行,银行帐号]。""" + return read_table(path, SUPPLIER_KEYS, SUPPLIER_ORDER) + + +@app.route("/import_template.xlsx") +@login_required +def import_template(): + import openpyxl + wb = openpyxl.Workbook() + ws = wb.active + ws.append([u"物料代码", u"物料规格", u"单位", u"数量", u"含税单价", u"交期", u"备注"]) + ws.append(["M-1001", u"铝合金支架 6061-T6", u"件", 100, 38.5, "2026-09-20", u""]) + for col, w in zip("ABCDEFG", (16, 34, 8, 10, 12, 14, 20)): + ws.column_dimensions[col].width = w + buf = io.BytesIO() + wb.save(buf) + buf.seek(0) + return send_file(buf, as_attachment=True, download_name=u"明细导入模板.xlsx", + mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet") + + +# ---------------- 供应商 / 物料 ---------------- +@app.route("/suppliers", methods=["GET", "POST"]) +@login_required +@role_required("admin", "maker") +def suppliers(): + u = current_user() + if request.method == "POST": + sid = request.form.get("id") + s = db.get(Supplier, int(sid)) if sid else Supplier() + for f in ("code", "name", "contact", "phone", "tax_no", "bank_name", + "bank_account", "payment_terms"): + setattr(s, f, request.form.get(f, "").strip()) + if not sid: + db.add(s) + db.commit() + flash(u"已保存", "ok") + return redirect(url_for("suppliers")) + rows = db.query(Supplier).order_by(Supplier.id).all() + return render_template("suppliers.html", user=u, rows=rows) + + +@app.route("/suppliers//toggle", methods=["POST"]) +@login_required +@role_required("admin", "maker") +def supplier_toggle(sid): + """停用 / 启用互切。停用后不出现在下单选择里，但保留历史数据。""" + s = db.get(Supplier, sid) + if s: + s.active = not s.active + db.commit() + flash(u"已%s：%s" % (u"启用" if s.active else u"停用", s.name), "ok") + return redirect(url_for("suppliers")) + + +@app.route("/suppliers//delete", methods=["POST"]) +@login_required +@role_required("admin") +def supplier_delete(sid): + """彻底删除。订单里存的是供应商名称快照，删库记录不影响历史单据。""" + s = db.get(Supplier, sid) + if s: + name = s.name + db.delete(s) + db.commit() + flash(u"已删除供应商：%s" % name, "ok") + return redirect(url_for("suppliers")) + + +@app.route("/suppliers/import_excel", methods=["POST"]) +@login_required +@role_required("admin", "maker") +def supplier_import_excel(): + file = request.files.get("file") + if not file or not file.filename: + flash(u"请选择文件", "error") + return redirect(url_for("suppliers")) + path = os.path.join(UPLOADS, "%s_%s" % (now().strftime("%H%M%S"), + secure_filename(file.filename))) + file.save(path) + try: + rows = read_supplier_rows(path) + except Exception as e: + flash(u"读取失败：%s" % e, "error") + return redirect(url_for("suppliers")) + mode = request.form.get("mode", "update") + added = updated = skipped = failed = 0 + errs = [] + + def s(v): + return str(v or "").strip() + + for i, r in enumerate(rows, start=2): + if not any(r): + continue + (code, name, contact, phone, payment_terms, tax_no, + bank_name, bank_account) = (list(r) + [""] * 8)[:8] + code, name = s(code), s(name) + if not name: + failed += 1 + errs.append(u"第%d行：供应商名称为空" % i) + continue + obj = db.query(Supplier).filter_by(code=code).first() if code else None + if not obj: + obj = db.query(Supplier).filter_by(name=name).first() + if obj: + if mode == "skip": + skipped += 1 + continue + obj.code = code or obj.code + obj.name = name + obj.contact = s(contact) + obj.phone = s(phone) + obj.payment_terms = s(payment_terms) + obj.tax_no = s(tax_no) + obj.bank_name = s(bank_name) + obj.bank_account = s(bank_account) + obj.active = True + updated += 1 + else: + db.add(Supplier(code=code, name=name, contact=s(contact), + phone=s(phone), payment_terms=s(payment_terms), + tax_no=s(tax_no), bank_name=s(bank_name), + bank_account=s(bank_account), active=True)) + added += 1 + db.commit() + msg = u"导入完成：新增 %d，更新 %d，跳过 %d，失败 %d" % (added, updated, skipped, failed) + if errs: + msg += u"。失败明细：" + u"；".join(errs[:8]) + if len(errs) > 8: + msg += u" 等共 %d 条" % len(errs) + flash(msg, "ok" if not failed else "error") + return redirect(url_for("suppliers")) + + +@app.route("/suppliers/import_template.xlsx") +@login_required +def supplier_import_template(): + import openpyxl + wb = openpyxl.Workbook() + ws = wb.active + ws.append([u"编号", u"名称", u"联系人", u"电话", u"付款方式", u"税号", + u"开户银行", u"银行帐号"]) + ws.append(["S003", u"苏州精锐五金制品有限公司", u"周经理", "13800000001", + u"月结30天", "91320500MA1XXXXXXX", u"中国银行苏州工业园区支行", + "4562 1088 0012 3456"]) + for col, w in zip("ABCDEFGH", (10, 30, 12, 16, 14, 22, 26, 24)): + ws.column_dimensions[col].width = w + buf = io.BytesIO() + wb.save(buf) + buf.seek(0) + return send_file(buf, as_attachment=True, download_name=u"供应商导入模板.xlsx", + mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet") + + +@app.route("/materials", methods=["GET", "POST"]) +@login_required +@role_required("admin", "maker") +def materials(): + u = current_user() + if request.method == "POST": + mid = request.form.get("id") + m = db.get(Material, int(mid)) if mid else Material() + m.code = request.form.get("code", "").strip() + m.spec = request.form.get("spec", "").strip() + m.unit = request.form.get("unit", "").strip() + try: + m.price = float(request.form.get("price") or 0) + except ValueError: + m.price = 0 + if not mid: + db.add(m) + db.commit() + flash(u"已保存", "ok") + return redirect(url_for("materials")) + rows = db.query(Material).order_by(Material.id).all() + return render_template("materials.html", user=u, rows=rows) + + +@app.route("/materials//toggle", methods=["POST"]) +@login_required +@role_required("admin", "maker") +def material_toggle(mid): + """停用 / 启用互切。停用后不出现在下单物料下拉里，但保留历史数据。""" + m = db.get(Material, mid) + if m: + m.active = not m.active + db.commit() + flash(u"已%s：%s" % (u"启用" if m.active else u"停用", m.code), "ok") + return redirect(url_for("materials")) + + +@app.route("/materials//delete", methods=["POST"]) +@login_required +@role_required("admin") +def material_delete(mid): + """彻底删除。订单明细里存的是物料快照，删库记录不影响历史单据。""" + m = db.get(Material, mid) + if m: + code = m.code + db.delete(m) + db.commit() + flash(u"已删除物料：%s" % code, "ok") + return redirect(url_for("materials")) + + +@app.route("/materials/import_excel", methods=["POST"]) +@login_required +@role_required("admin", "maker") +def material_import_excel(): + file = request.files.get("file") + if not file or not file.filename: + flash(u"请选择文件", "error") + return redirect(url_for("materials")) + path = os.path.join(UPLOADS, "%s_%s" % (now().strftime("%H%M%S"), + secure_filename(file.filename))) + file.save(path) + try: + rows = read_material_rows(path) + except Exception as e: + flash(u"读取失败：%s" % e, "error") + return redirect(url_for("materials")) + mode = request.form.get("mode", "update") + added = updated = skipped = failed = 0 + errs = [] + + def s(v): + return str(v or "").strip() + + def num(v): + try: + return float(str(v or "").strip() or 0) + except (TypeError, ValueError): + return 0.0 + + for i, r in enumerate(rows, start=2): + if not any(r): + continue + code, spec, unit, price = (list(r) + [""] * 4)[:4] + code = s(code) + if not code: + failed += 1 + errs.append(u"第%d行：物料代码为空" % i) + continue + m = db.query(Material).filter_by(code=code).first() + if m: + if mode == "skip": + skipped += 1 + continue + m.spec = s(spec) + m.unit = s(unit) + m.price = num(price) + m.active = True + updated += 1 + else: + db.add(Material(code=code, spec=s(spec), unit=s(unit), + price=num(price), active=True)) + added += 1 + db.commit() + msg = u"导入完成：新增 %d，更新 %d，跳过 %d，失败 %d" % (added, updated, skipped, failed) + if errs: + msg += u"。失败明细：" + u"；".join(errs[:8]) + if len(errs) > 8: + msg += u" 等共 %d 条" % len(errs) + flash(msg, "ok" if not failed else "error") + return redirect(url_for("materials")) + + +@app.route("/materials/import_template.xlsx") +@login_required +def material_import_template(): + import openpyxl + wb = openpyxl.Workbook() + ws = wb.active + ws.append([u"物料代码", u"物料规格", u"单位", u"默认含税单价"]) + ws.append(["M-1004", u"铝合金支架 6061-T6 200×80×10mm", u"件", 38.5]) + ws.append(["M-1005", u"步进电机 57BYG250 2相", u"台", 265.0]) + for col, w in zip("ABCD", (16, 34, 10, 16)): + ws.column_dimensions[col].width = w + buf = io.BytesIO() + wb.save(buf) + buf.seek(0) + return send_file(buf, as_attachment=True, download_name=u"物料导入模板.xlsx", + mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet") + + +@app.route("/api/materials") +@login_required +def api_materials(): + rows = db.query(Material).filter_by(active=True).all() + return jsonify([{"id": m.id, "code": m.code, "spec": m.spec, + "unit": m.unit, "price": m.price} for m in rows]) + + +@app.route("/api/suppliers") +@login_required +def api_suppliers(): + rows = db.query(Supplier).filter_by(active=True).all() + return jsonify([{"id": s.id, "name": s.name, "contact": s.contact, + "phone": s.phone, "tax_no": s.tax_no, "bank_name": s.bank_name, + "bank_account": s.bank_account, + "payment_terms": s.payment_terms} for s in rows]) + + +# ---------------- 用户 ---------------- +@app.route("/users", methods=["GET", "POST"]) +@login_required +@role_required("admin") +def users(): + u = current_user() + if request.method == "POST": + uid = request.form.get("id") + t = db.get(User, int(uid)) if uid else User() + t.username = request.form.get("username", "").strip() + t.name = request.form.get("name", "").strip() + t.role = request.form.get("role", "maker") + pw = request.form.get("password", "").strip() + if pw: + t.password_hash = generate_password_hash(pw) + elif not uid: + t.password_hash = generate_password_hash("123456") + if not uid: + db.add(t) + db.commit() + flash(u"已保存", "ok") + return redirect(url_for("users")) + rows = db.query(User).order_by(User.id).all() + return render_template("users.html", user=u, rows=rows, roles=ROLE_TEXT) + + +# ---------------- 统计 ---------------- +@app.route("/stats") +@login_required +def stats(): + u = current_user() + by_status = (db.query(Order.status, func.count(Order.id), func.sum(Order.total_amount)) + .group_by(Order.status).all()) + by_supplier = (db.query(Order.supplier_name, func.count(Order.id), + func.sum(Order.total_amount)) + .filter(Order.status == "approved") + .group_by(Order.supplier_name) + .order_by(func.sum(Order.total_amount).desc()).limit(20).all()) + by_month = (db.query(func.substr(Order.order_date, 1, 7), func.count(Order.id), + func.sum(Order.total_amount)) + .group_by(func.substr(Order.order_date, 1, 7)).all()) + return render_template("stats.html", user=u, by_status=by_status, + by_supplier=by_supplier, by_month=by_month, + status_text=STATUS_TEXT) + + +@app.route("/export/ledger") +@login_required +def export_ledger(): + import openpyxl + rows = db.query(Order).order_by(Order.id.desc()).all() + wb = openpyxl.Workbook() + ws = wb.active + ws.append([u"采购单号", u"单据类型", u"供应商", u"采购日期", u"状态", u"金额", + u"制单人", u"审核人", u"批准人", u"创建时间"]) + for o in rows: + ws.append([o.order_no, o.doc_type_text, o.supplier_name, o.order_date, + o.status_text, o.total_amount, o.created_by_name, + o.reviewed_by, o.approved_by, + o.created_at.strftime("%Y-%m-%d %H:%M") if o.created_at else ""]) + for col, w in zip("ABCDEFGHIJ", (18, 14, 30, 12, 12, 12, 12, 12, 12, 18)): + ws.column_dimensions[col].width = w + buf = io.BytesIO() + wb.save(buf) + buf.seek(0) + return send_file(buf, as_attachment=True, + download_name=u"采购订单台账_%s.xlsx" % now().strftime("%Y%m%d"), + mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet") + + +@app.route("/exports/") +@login_required +def exports_file(name): + return send_from_directory(EXPORTS, name) + + +@app.context_processor +def inject_common(): + return {"COMPANY_NAME": COMPANY_NAME, "STATUS_TEXT": STATUS_TEXT, + "ROLE_TEXT": ROLE_TEXT, "DOC_TYPES": DOC_TYPES} + + +def port_busy(p): + import socket + s = socket.socket() + s.settimeout(0.4) + try: + s.connect(("127.0.0.1", p)) + return True + except Exception: + return False + finally: + s.close() + + +if __name__ == "__main__": + import threading + import webbrowser + + port = int(sys.argv[1]) if len(sys.argv) > 1 else 8080 + while port_busy(port) and port < 8120: + port += 1 + if getattr(sys, "frozen", False) or os.environ.get("PO_OPEN_BROWSER") == "1": + threading.Timer(1.5, lambda: webbrowser.open("http://127.0.0.1:%d" % port)).start() + print("=" * 56) + print(" 博特维采购订单管理系统已启动") + print(" 本机访问： http://127.0.0.1:%d" % port) + print(" 关闭本窗口即停止服务") + print("=" * 56) + app.run(host="0.0.0.0", port=port, debug=False, threaded=True) diff --git a/po_app/models.py b/po_app/models.py new file mode 100644 index 0000000..397566c --- /dev/null +++ b/po_app/models.py @@ -0,0 +1,154 @@ +# -*- coding: utf-8 -*- +"""数据模型：用户、供应商、物料、订单、明细、审核记录。""" +from datetime import datetime + +from sqlalchemy import (Boolean, Column, DateTime, Float, ForeignKey, Integer, + String, Text, create_engine) +from sqlalchemy.orm import declarative_base, relationship, sessionmaker + +Base = declarative_base() + +COMPANY_NAME = u"苏州博特维智能科技有限公司" +DEFAULT_DELIVERY_PLACE = (u"苏州市工业园区星汉街5号腾飞新苏工业坊B幢5楼05-06单元 " + u"胡乃琛 13913172516") +# 需方（我方）默认信息：新建单据时自动预置到联系人 / 电话 / 税号 / 税点 +DEFAULT_OUR_CONTACT = u"胡乃琛" +DEFAULT_OUR_PHONE = u"13913172516" +DEFAULT_OUR_TAX_NO = u"91320507MA1WW8D15D" +DEFAULT_TAX_RATE = u"13%" + +DOC_TYPES = { + "purchase": u"采购单", + "outsourcing": u"委外生产订单", +} + +STATUS_TEXT = { + "draft": u"草稿", + "pending_review": u"待主管审核", + "pending_approval": u"待经理批准", + "approved": u"已批准", + "rejected": u"已驳回", +} + +ROLE_TEXT = { + "admin": u"系统管理员", + "maker": u"制单员", + "reviewer": u"主管审核", + "approver": u"经理批准", +} + + +class User(Base): + __tablename__ = "users" + id = Column(Integer, primary_key=True) + username = Column(String(50), unique=True, nullable=False) + password_hash = Column(String(200), nullable=False) + name = Column(String(50), nullable=False) + role = Column(String(20), nullable=False, default="maker") + active = Column(Boolean, default=True) + created_at = Column(DateTime, default=datetime.now) + + +class Supplier(Base): + __tablename__ = "suppliers" + id = Column(Integer, primary_key=True) + code = Column(String(50)) + name = Column(String(200), nullable=False) + contact = Column(String(50)) + phone = Column(String(50)) + tax_no = Column(String(80)) + bank_name = Column(String(120)) + bank_account = Column(String(80)) + payment_terms = Column(String(120)) + active = Column(Boolean, default=True) + + +class Material(Base): + __tablename__ = "materials" + id = Column(Integer, primary_key=True) + code = Column(String(80), nullable=False) + spec = Column(String(300)) + unit = Column(String(20)) + price = Column(Float, default=0) + active = Column(Boolean, default=True) + + +class Order(Base): + __tablename__ = "orders" + id = Column(Integer, primary_key=True) + order_no = Column(String(50), nullable=False) + doc_type = Column(String(20), nullable=False, default="purchase") + + supplier_name = Column(String(200)) + supplier_contact = Column(String(50)) + supplier_phone = Column(String(50)) + supplier_tax_no = Column(String(80)) + supplier_bank_name = Column(String(120)) + supplier_bank_account = Column(String(80)) + payment_terms = Column(String(120)) + + our_company = Column(String(200), default=COMPANY_NAME) + our_contact = Column(String(50), default=DEFAULT_OUR_CONTACT) + our_phone = Column(String(50), default=DEFAULT_OUR_PHONE) + our_tax_no = Column(String(80), default=DEFAULT_OUR_TAX_NO) + tax_rate = Column(String(30), default=DEFAULT_TAX_RATE) + order_date = Column(String(20)) + delivery_place = Column(String(300), default=DEFAULT_DELIVERY_PLACE) + note = Column(Text) + + total_amount = Column(Float, default=0) + + status = Column(String(20), default="draft") + created_by = Column(Integer, ForeignKey("users.id")) + created_by_name = Column(String(50)) + created_at = Column(DateTime, default=datetime.now) + submitted_at = Column(DateTime) + reviewed_by = Column(String(50)) + reviewed_at = Column(DateTime) + approved_by = Column(String(50)) + approved_at = Column(DateTime) + reject_reason = Column(Text) + + items = relationship("OrderItem", backref="order", cascade="all, delete-orphan", + order_by="OrderItem.seq") + logs = relationship("ApprovalLog", backref="order", cascade="all, delete-orphan", + order_by="ApprovalLog.id") + + @property + def status_text(self): + return STATUS_TEXT.get(self.status, self.status) + + @property + def doc_type_text(self): + return DOC_TYPES.get(self.doc_type, self.doc_type) + + +class OrderItem(Base): + __tablename__ = "order_items" + id = Column(Integer, primary_key=True) + order_id = Column(Integer, ForeignKey("orders.id")) + seq = Column(Integer, default=1) + material_code = Column(String(80)) + material_spec = Column(String(300)) + unit = Column(String(20)) + qty = Column(Float, default=0) + price = Column(Float, default=0) + amount = Column(Float, default=0) + delivery_date = Column(String(30)) + remark = Column(String(200)) + + +class ApprovalLog(Base): + __tablename__ = "approval_logs" + id = Column(Integer, primary_key=True) + order_id = Column(Integer, ForeignKey("orders.id")) + action = Column(String(50)) + user_name = Column(String(50)) + comment = Column(Text) + created_at = Column(DateTime, default=datetime.now) + + +def init_db(db_path): + engine = create_engine("sqlite:///" + db_path, echo=False) + Base.metadata.create_all(engine) + return engine, sessionmaker(bind=engine) diff --git a/po_app/pdf_export.py b/po_app/pdf_export.py new file mode 100644 index 0000000..a00efd0 --- /dev/null +++ b/po_app/pdf_export.py @@ -0,0 +1,397 @@ +# -*- coding: utf-8 -*- +"""按原 Excel 模板版式生成采购单 / 委外生产订单 PDF。""" +import math +import os +from datetime import datetime +from xml.sax.saxutils import escape + +from reportlab.lib import colors +from reportlab.lib.enums import TA_CENTER, TA_LEFT +from reportlab.lib.pagesizes import A4 +from reportlab.lib.styles import ParagraphStyle +from reportlab.lib.units import mm +from reportlab.pdfbase import pdfmetrics +from reportlab.pdfbase.ttfonts import TTFont +from reportlab.platypus import (BaseDocTemplate, Flowable, Frame, KeepTogether, + LongTable, PageTemplate, Paragraph, Spacer, + Table, TableStyle) + +from amount_cn import amount_to_cn +from models import COMPANY_NAME, DOC_TYPES + +FONT_DIRS = [r"C:\Windows\Fonts", "/usr/share/fonts"] +BODY_CANDIDATES = ["simsun.ttc", "STSONG.TTF", "simfang.ttf", "simkai.ttf", "Deng.ttf"] +BOLD_CANDIDATES = ["simhei.ttf", "msyh.ttc", "msyhbd.ttc", "Dengb.ttf", "STXIHEI.TTF"] + +FONT_BODY = "CN-Body" +FONT_BOLD = "CN-Bold" + +# 明细列宽比例，取自两个 xls 模板的实际列宽 +COL_RATIO = { + "purchase": [1184, 3104, 6528, 3232, 2400, 2976, 2400, 3712, 1728], + "outsourcing": [1184, 5440, 3392, 5408, 2400, 2976, 3840, 3712, 4736], +} + +TERMS = [ + u"1、因供方交期延误，品质不良，规格不符或数量不准确等情况造成我公司损失，概由供应方承担。", + u"2、供方所交材料均要求符合欧盟环境有害物质管理标准。", + u"3、随货请附出货检验报告。", + u"4、运输与保险：供方负责。", + u"5、违约责任：按经济合同法等相关法律法规处置，供方延期交货，需方有权延期付款，延期付款时间为延期交货时间的2倍。", + u"6、保密协定：供需双方保证对方的商业秘密。", + u"7、合同生效：本合同一式两份，自双方签字盖章起生效，传真件签字盖章亦有效，订单手写无效。", + u"8、售后服务与保修：如产品使用过程中发现质量问题，供方负责调换或退货，并赔偿相应损失，包含但不限于物料和返工费用。", +] + +STATUS_WATERMARK = { + "draft": u"草稿 · 未提交", + "pending_review": u"待主管审核", + "pending_approval": u"待经理批准", + "rejected": u"已驳回 · 未生效", +} + + +SEAL_COLOR = colors.HexColor("#C8102E") + + +def _arc_text(c, text, cx, cy, radius, start_deg, spread_deg, font, size): + """沿圆弧排字，用于印章上下缘。角度以度为单位，逆时针为正。""" + if not text: + return + c.setFont(font, size) + n = len(text) + step = spread_deg / float(n - 1) if n > 1 else 0.0 + for i, ch in enumerate(text): + ang = start_deg - i * step + rad = math.radians(ang) + c.saveState() + c.translate(cx + radius * math.cos(rad), cy + radius * math.sin(rad)) + c.rotate(ang - 90) + c.drawCentredString(0, 0, ch) + c.restoreState() + + +def _star(c, cx, cy, r_out, points=5, inner_ratio=0.382): + """画五角星（填充）。""" + path = c.beginPath() + for i in range(points * 2): + r = r_out if i % 2 == 0 else r_out * inner_ratio + ang = math.radians(90 + i * (180.0 / points)) + x = cx + r * math.cos(ang) + y = cy + r * math.sin(ang) + if i == 0: + path.moveTo(x, y) + else: + path.lineTo(x, y) + path.close() + c.drawPath(path, stroke=0, fill=1) + + +class SealFlowable(Flowable): + """红色圆形电子章：外圈 + 上缘公司名 + 五角星 + 中部用途 + 下缘审核人/日期。""" + + def __init__(self, company, title, who, when, size=28 * mm): + Flowable.__init__(self) + self.company = company or u"" + self.title = title or u"审核专用章" + self.who = who or u"" + self.when = when or u"" + self.width = size + self.height = size + + def draw(self): + c = self.canv + w, h = self.width, self.height + cx, cy = w / 2.0, h / 2.0 + r = min(w, h) / 2.0 - 1.2 * mm + c.saveState() + c.setFillColor(SEAL_COLOR) + c.setStrokeColor(SEAL_COLOR) + c.setLineWidth(1.5) + c.circle(cx, cy, r, stroke=1, fill=0) + c.setLineWidth(0.5) + c.circle(cx, cy, r - 1.1 * mm, stroke=1, fill=0) + # 上缘公司名 + _arc_text(c, self.company, cx, cy, r - 3.2 * mm, 152, -124, FONT_BODY, 5.8) + # 中心五角星 + _star(c, cx, cy + 2.4 * mm, 2.8 * mm) + # 中部用途 + c.setFont(FONT_BOLD, 8.0) + c.drawCentredString(cx, cy - 4.2 * mm, self.title) + # 下缘审核人 + 日期 + line = u" ".join(x for x in (self.who, self.when) if x) + if line: + c.setFont(FONT_BODY, 5.4) + c.drawCentredString(cx, cy - 9.0 * mm, line) + c.restoreState() + + +def _find_font(candidates): + for d in FONT_DIRS: + for name in candidates: + p = os.path.join(d, name) + if os.path.exists(p): + return p + return None + + +def register_fonts(): + if FONT_BODY in pdfmetrics.getRegisteredFontNames(): + return + body = _find_font(BODY_CANDIDATES) + bold = _find_font(BOLD_CANDIDATES) + if not body: + raise RuntimeError(u"未找到中文字体，请确认系统已安装宋体或黑体") + pdfmetrics.registerFont(TTFont(FONT_BODY, body, subfontIndex=0)) + pdfmetrics.registerFont(TTFont(FONT_BOLD, bold or body, subfontIndex=0)) + + +def _fmt_money(v): + try: + return "{:,.2f}".format(float(v or 0)) + except (TypeError, ValueError): + return "0.00" + + +def _fmt_num(v): + try: + return "{:g}".format(round(float(v or 0), 4)) + except (TypeError, ValueError): + return "" + + +def P(text, style): + """转义后生成 Paragraph，避免 & < > 破坏 XML。""" + s = escape(u"" if text is None else str(text)).replace("\n", "
") + return Paragraph(s or " ", style) + + +def build_pdf(order, items, out_path, watermark=None): + register_fonts() + if watermark is None: + watermark = STATUS_WATERMARK.get(order.status) + + page_w, page_h = A4 + margin_x = 11 * mm + doc = BaseDocTemplate(out_path, pagesize=A4, + leftMargin=margin_x, rightMargin=margin_x, + topMargin=11 * mm, bottomMargin=16 * mm, + title=u"%s %s" % (order.doc_type_text, order.order_no), + author=COMPANY_NAME) + frame = Frame(margin_x, 16 * mm, page_w - 2 * margin_x, + page_h - 27 * mm, id="body", showBoundary=0) + + counters = {"total": 0} + + def on_page(canvas, doc_): + canvas.saveState() + canvas.setFont(FONT_BODY, 8) + canvas.setFillColor(colors.HexColor("#666666")) + footer = u"%s %s 第 %d 页 / 共 %d 页 打印时间 %s" % ( + order.doc_type_text, order.order_no, doc_.page, + counters["total"] or doc_.page, + datetime.now().strftime("%Y-%m-%d %H:%M")) + canvas.drawCentredString(page_w / 2.0, 10 * mm, footer) + canvas.setStrokeColor(colors.HexColor("#BBBBBB")) + canvas.setLineWidth(0.4) + canvas.line(margin_x, 13 * mm, page_w - margin_x, 13 * mm) + if watermark: + canvas.setFont(FONT_BOLD, 34) + canvas.setFillColor(colors.Color(0.85, 0.15, 0.15, alpha=0.15)) + canvas.saveState() + canvas.translate(page_w / 2.0, page_h / 2.0) + canvas.rotate(28) + canvas.drawCentredString(0, 0, watermark) + canvas.restoreState() + canvas.restoreState() + + doc.addPageTemplates([PageTemplate(id="main", frames=[frame], onPage=on_page)]) + + styles = { + "title": ParagraphStyle("title", fontName=FONT_BOLD, fontSize=17, leading=22, + alignment=TA_CENTER), + "company": ParagraphStyle("company", fontName=FONT_BOLD, fontSize=12.5, leading=17, + alignment=TA_CENTER), + "cell": ParagraphStyle("cell", fontName=FONT_BODY, fontSize=8.6, leading=11.5, + alignment=TA_LEFT), + "cell_c": ParagraphStyle("cell_c", fontName=FONT_BODY, fontSize=8.6, leading=11.5, + alignment=TA_CENTER), + "th": ParagraphStyle("th", fontName=FONT_BOLD, fontSize=9, leading=12, + alignment=TA_CENTER), + "term": ParagraphStyle("term", fontName=FONT_BODY, fontSize=8.8, leading=13.5), + "sign": ParagraphStyle("sign", fontName=FONT_BODY, fontSize=9.5, leading=15), + } + + def make_story(): + story = [] + story.append(P(COMPANY_NAME, styles["company"])) + story.append(Spacer(1, 3 * mm)) + story.append(P(order.doc_type_text or DOC_TYPES.get(order.doc_type, u""), + styles["title"])) + story.append(Spacer(1, 4 * mm)) + + # ---------- 抬头区 ---------- + left_rows = [ + (u"供应商：", order.supplier_name), + (u"联系人：", order.supplier_contact), + (u"电 话：", order.supplier_phone), + (u"付款方式：", order.payment_terms), + (u"税 号：", order.supplier_tax_no), + (u"银行帐号：", order.supplier_bank_account), + (u"开户银行：", order.supplier_bank_name), + ] + right_rows = [ + (u"需 方：", order.our_company), + (u"联系人：", order.our_contact), + (u"电 话：", order.our_phone), + (u"税 点：", order.tax_rate), + (u"税 号：", order.our_tax_no), + (u"采购单号：", order.order_no), + (u"采购日期：", order.order_date), + ] + head_data = [[P(l1 + (v1 or ""), styles["cell"]), P(l2 + (v2 or ""), styles["cell"])] + for (l1, v1), (l2, v2) in zip(left_rows, right_rows)] + head = Table(head_data, colWidths=[93 * mm, 93 * mm], rowHeights=[6.6 * mm] * 7) + head.setStyle(TableStyle([ + ("GRID", (0, 0), (-1, -1), 0.5, colors.HexColor("#333333")), + ("VALIGN", (0, 0), (-1, -1), "MIDDLE"), + ("LEFTPADDING", (0, 0), (-1, -1), 4), + ("RIGHTPADDING", (0, 0), (-1, -1), 4), + ("TOPPADDING", (0, 0), (-1, -1), 1.5), + ("BOTTOMPADDING", (0, 0), (-1, -1), 1.5), + ])) + story.append(head) + story.append(Spacer(1, 4 * mm)) + + # ---------- 明细表 ---------- + ratio = COL_RATIO.get(order.doc_type, COL_RATIO["purchase"]) + total_w = 186 * mm + s = float(sum(ratio)) + col_w = [total_w * r / s for r in ratio] + # 序号列保底宽度，避免表头换行；差值从最宽的规格列扣除 + min_first = 9 * mm + if col_w[0] < min_first: + col_w[2] -= (min_first - col_w[0]) + col_w[0] = min_first + headers = [u"序号", u"物料代码", u"物料规格", u"单位", u"数量", u"含税单价", + u"金额", u"交期", u"备注"] + data = [[P(h, styles["th"]) for h in headers]] + total = 0.0 + for idx, it in enumerate(items, 1): + total += float(it.amount or 0) + data.append([ + P(str(idx), styles["cell_c"]), + P(it.material_code, styles["cell"]), + P(it.material_spec, styles["cell"]), + P(it.unit, styles["cell_c"]), + P(_fmt_num(it.qty), styles["cell_c"]), + P(_fmt_money(it.price), styles["cell_c"]), + P(_fmt_money(it.amount), styles["cell_c"]), + P(it.delivery_date, styles["cell_c"]), + P(it.remark, styles["cell"]), + ]) + while len(data) - 1 < 5: + data.append([P("", styles["cell"]) for _ in range(9)]) + detail = LongTable(data, colWidths=col_w, repeatRows=1) + detail.setStyle(TableStyle([ + ("GRID", (0, 0), (-1, -1), 0.5, colors.HexColor("#333333")), + ("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#EFEFEF")), + ("VALIGN", (0, 0), (-1, -1), "MIDDLE"), + ("LEFTPADDING", (0, 0), (-1, -1), 3), + ("RIGHTPADDING", (0, 0), (-1, -1), 3), + ("TOPPADDING", (0, 0), (-1, -1), 3), + ("BOTTOMPADDING", (0, 0), (-1, -1), 3), + ])) + story.append(detail) + + # ---------- 合计区 ---------- + cn_total = amount_to_cn(total) + sum_data = [ + [P(u"合计（小写）", styles["cell"]), P("", styles["cell"]), P("", styles["cell"]), + P(_fmt_money(total), styles["cell_c"])] + [P("", styles["cell"]) for _ in range(5)], + [P(u"合计（人民币)", styles["cell"]), P("", styles["cell"]), + P(cn_total, styles["cell_c"])] + [P("", styles["cell"]) for _ in range(4)] + + [P(u"元整", styles["cell"]), P("", styles["cell"])], + ] + sum_t = Table(sum_data, colWidths=col_w, rowHeights=[7.5 * mm, 7.5 * mm]) + sum_t.setStyle(TableStyle([ + ("GRID", (0, 0), (-1, -1), 0.5, colors.HexColor("#333333")), + ("SPAN", (0, 0), (2, 0)), + ("SPAN", (3, 0), (8, 0)), + ("SPAN", (0, 1), (1, 1)), + ("SPAN", (2, 1), (6, 1)), + ("SPAN", (7, 1), (8, 1)), + ("VALIGN", (0, 0), (-1, -1), "MIDDLE"), + ("LEFTPADDING", (0, 0), (-1, -1), 4), + ("RIGHTPADDING", (0, 0), (-1, -1), 4), + ])) + story.append(sum_t) + story.append(Spacer(1, 3 * mm)) + + # ---------- 备注条款 ---------- + terms_block = [P(u"备注：", styles["term"])] + for t in TERMS: + terms_block.append(P(t, styles["term"])) + if order.delivery_place: + terms_block.append(P(u"9、交货地点：%s" % order.delivery_place, styles["term"])) + if order.note: + terms_block.append(P(u"补充说明：%s" % order.note, styles["term"])) + story.append(KeepTogether(terms_block)) + story.append(Spacer(1, 8 * mm)) + + # ---------- 签署区 ---------- + def stamp(name, dt): + return (u"%s %s" % (name, dt.strftime("%Y-%m-%d %H:%M"))) if dt else (name or u"待处理") + + sign_line = u"制单：%s 主管审核：%s 经理批准：%s" % ( + stamp(order.created_by_name, order.created_at), + stamp(order.reviewed_by, order.reviewed_at), + stamp(order.approved_by, order.approved_at)) + confirm_line = (u"供方确认（盖章）：__________________ " + u"需方确认（盖章）：__________________") + # 已审核 / 已批准 → 在签署区右侧盖章（两枚时并排） + seals = [] + if order.reviewed_by: + seals.append(SealFlowable( + order.our_company or COMPANY_NAME, u"审核专用章", order.reviewed_by, + order.reviewed_at.strftime("%Y-%m-%d") if order.reviewed_at else u"")) + if order.approved_by: + seals.append(SealFlowable( + order.our_company or COMPANY_NAME, u"批准专用章", order.approved_by, + order.approved_at.strftime("%Y-%m-%d") if order.approved_at else u"")) + if seals: + if len(seals) == 1: + seal_cell = seals[0] + else: + inner = Table([seals], colWidths=[29 * mm, 29 * mm]) + inner.setStyle(TableStyle([ + ("VALIGN", (0, 0), (-1, -1), "CENTER"), + ("LEFTPADDING", (0, 0), (-1, -1), 0), + ("RIGHTPADDING", (0, 0), (-1, -1), 0), + ("TOPPADDING", (0, 0), (-1, -1), 0), + ("BOTTOMPADDING", (0, 0), (-1, -1), 0), + ])) + seal_cell = inner + sign_tbl = Table( + [[P(sign_line, styles["sign"]), seal_cell], + [P(confirm_line, styles["sign"]), P("", styles["sign"])]], + colWidths=[128 * mm, 58 * mm]) + sign_tbl.setStyle(TableStyle([ + ("VALIGN", (0, 0), (0, -1), "TOP"), + ("VALIGN", (1, 0), (1, -1), "CENTER"), + ("LEFTPADDING", (0, 0), (-1, -1), 0), + ("RIGHTPADDING", (0, 0), (-1, -1), 0), + ("TOPPADDING", (0, 0), (-1, -1), 0), + ("BOTTOMPADDING", (0, 0), (-1, -1), 0), + ])) + story.append(KeepTogether(sign_tbl)) + else: + story.append(P(sign_line, styles["sign"])) + story.append(Spacer(1, 3 * mm)) + story.append(P(confirm_line, styles["sign"])) + return story + + doc.build(make_story()) + counters["total"] = doc.page + doc.build(make_story()) + return out_path diff --git a/po_app/requirements.txt b/po_app/requirements.txt new file mode 100644 index 0000000..e4b5ec0 --- /dev/null +++ b/po_app/requirements.txt @@ -0,0 +1,5 @@ +Flask>=3.0 +SQLAlchemy>=2.0 +reportlab>=4.0 +openpyxl>=3.1 +xlrd>=2.0 diff --git a/po_app/static/style.css b/po_app/static/style.css new file mode 100644 index 0000000..faeaa0a --- /dev/null +++ b/po_app/static/style.css @@ -0,0 +1,121 @@ +* { box-sizing: border-box; } +body { + margin: 0; + font-family: "Microsoft YaHei", "SimHei", "PingFang SC", sans-serif; + background: #f4f6f8; + color: #222; + font-size: 14px; +} +a { color: #1668dc; text-decoration: none; } +a:hover { text-decoration: underline; } +.topbar { + background: #123a6b; color: #fff; padding: 0 20px; height: 52px; + display: flex; align-items: center; gap: 22px; +} +.topbar .brand { font-size: 16px; font-weight: 600; letter-spacing: .5px; } +.topbar nav a { color: #d8e6f7; font-size: 14px; } +.topbar nav a:hover, .topbar nav a.active { color: #fff; text-decoration: none; } +.topbar .spacer { flex: 1; } +.topbar .me { color: #cfe0f5; font-size: 13px; } +.wrap { max-width: 1180px; margin: 18px auto 40px; padding: 0 16px; } +.card { + background: #fff; border: 1px solid #e2e6ea; border-radius: 6px; + padding: 16px 18px; margin-bottom: 16px; +} +.card h2 { margin: 0 0 12px; font-size: 15px; font-weight: 600; } +.card h3 { margin: 18px 0 8px; font-size: 14px; font-weight: 600; color: #444; } +h1.page { font-size: 18px; margin: 4px 0 14px; } +table.grid { width: 100%; border-collapse: collapse; font-size: 13px; } +table.grid th, table.grid td { + border: 1px solid #dfe3e8; padding: 6px 8px; text-align: left; vertical-align: middle; +} +table.grid th { background: #f0f3f7; font-weight: 600; white-space: nowrap; } +table.grid tr:hover td { background: #fafbfc; } +table.grid td.num, table.grid th.num { text-align: right; } +.form-grid { display: grid; grid-template-columns: repeat(3, 1fr); gap: 10px 16px; } +.form-grid.two { grid-template-columns: repeat(2, 1fr); } +label { display: block; font-size: 12px; color: #666; margin-bottom: 3px; } +input[type=text], input[type=password], input[type=number], input[type=date], select, textarea { + width: 100%; padding: 6px 8px; border: 1px solid #ccd3da; border-radius: 4px; + font-size: 13px; font-family: inherit; background: #fff; +} +textarea { min-height: 60px; resize: vertical; } +input:focus, select:focus, textarea:focus { outline: none; border-color: #1668dc; } +.btn { + display: inline-block; padding: 7px 14px; border: 1px solid #c4ccd4; border-radius: 4px; + background: #fff; color: #333; font-size: 13px; cursor: pointer; font-family: inherit; +} +.btn:hover { background: #f2f5f8; text-decoration: none; } +.btn.primary { background: #1668dc; border-color: #1668dc; color: #fff; } +.btn.primary:hover { background: #0f57bd; } +.btn.ok { background: #12a05c; border-color: #12a05c; color: #fff; } +.btn.ok:hover { background: #0e8a4f; } +.btn.danger { background: #d9363e; border-color: #d9363e; color: #fff; } +.btn.danger:hover { background: #bc2b32; } +.btn.sm { padding: 4px 9px; font-size: 12px; } +.toolbar { display: flex; gap: 8px; flex-wrap: wrap; align-items: center; margin-bottom: 12px; } +.toolbar .grow { flex: 1; } +.flash { padding: 9px 12px; border-radius: 4px; margin-bottom: 12px; font-size: 13px; } +.flash.ok { background: #e8f6ee; border: 1px solid #b7e2cb; color: #12703f; } +.flash.error { background: #fdecec; border: 1px solid #f4bcbc; color: #a32026; } +.badge { display: inline-block; padding: 2px 8px; border-radius: 10px; font-size: 12px; } +.badge.draft { background: #eef0f2; color: #666; } +.badge.pending_review { background: #fff4e0; color: #a35c00; } +.badge.pending_approval { background: #e6f0ff; color: #14509b; } +.badge.approved { background: #e8f6ee; color: #12703f; } +.badge.rejected { background: #fdecec; color: #a32026; } +.stat-row { display: grid; grid-template-columns: repeat(5, 1fr); gap: 12px; margin-bottom: 16px; } +.stat { background: #fff; border: 1px solid #e2e6ea; border-radius: 6px; padding: 12px 14px; } +.stat .v { font-size: 22px; font-weight: 600; } +.stat .k { font-size: 12px; color: #777; margin-top: 2px; } +.muted { color: #888; font-size: 12px; } +.right { text-align: right; } +.center { text-align: center; } +.nowrap { white-space: nowrap; } +.login-box { width: 340px; margin: 90px auto; background: #fff; padding: 24px; + border: 1px solid #e2e6ea; border-radius: 8px; } +.login-box h2 { margin: 0 0 16px; font-size: 17px; text-align: center; } +.login-box .field { margin-bottom: 12px; } +.timeline { list-style: none; margin: 0; padding: 0; font-size: 13px; } +.timeline li { padding: 6px 0; border-bottom: 1px dashed #e5e8eb; } +.timeline li:last-child { border-bottom: none; } +.kv-table td { padding: 5px 8px; border: 1px solid #dfe3e8; font-size: 13px; } +.kv-table td.k { background: #f5f7f9; color: #555; width: 110px; white-space: nowrap; } +.total-bar { display: flex; gap: 18px; align-items: center; margin: 10px 0; font-size: 14px; } +.total-bar .amt { font-size: 18px; font-weight: 600; color: #c0392b; } +.hint { background: #f7f9fb; border-left: 3px solid #1668dc; padding: 8px 12px; + font-size: 12.5px; color: #555; margin-bottom: 12px; } +footer.foot { text-align: center; color: #999; font-size: 12px; margin: 24px 0; } +/* 卡片标题栏：标题 + 右侧操作按钮 */ +.card-head { display: flex; align-items: center; justify-content: space-between; + gap: 12px; margin-bottom: 12px; } +.card-head h2 { margin: 0; } +.field { margin-bottom: 12px; } +/* 模态弹窗 */ +.modal-mask { position: fixed; top: 0; left: 0; right: 0; bottom: 0; + background: rgba(0,0,0,.45); z-index: 1000; + display: flex; align-items: center; justify-content: center; } +.modal-box { background: #fff; border-radius: 8px; width: 520px; max-width: 92vw; + max-height: 88vh; overflow: auto; box-shadow: 0 8px 32px rgba(0,0,0,.25); } +.modal-head { padding: 12px 16px; border-bottom: 1px solid #e5e8eb; font-size: 15px; + font-weight: 600; display: flex; align-items: center; justify-content: space-between; } +.modal-close { cursor: pointer; color: #888; font-size: 22px; line-height: 1; padding: 0 4px; } +.modal-close:hover { color: #333; } +.modal-body { padding: 16px; } +.modal-foot { padding: 12px 16px; border-top: 1px solid #e5e8eb; + display: flex; justify-content: flex-end; gap: 8px; } +/* 停用记录：整行置灰 */ +table.grid tr.row-off td, +table.grid tr.row-off:hover td { color: #a3a8ad; background: #f5f6f7; } +table.grid tr.row-off td a, table.grid tr.row-off td b { color: #a3a8ad; } +table.grid tr.row-off .btn { opacity: .8; } +/* 电子章预览（页面上的红色圆形章） */ +.seal-wrap { display: inline-block; padding: 6px 10px; border: 1px dashed #d9b3b3; + border-radius: 6px; background: #fffafa; } +.seal-tip { color: #12703f; font-size: 13px; font-weight: 600; margin-bottom: 6px; } +.seal-circle { width: 126px; height: 126px; border: 3px solid #C8102E; border-radius: 50%; + color: #C8102E; display: flex; flex-direction: column; align-items: center; + justify-content: center; text-align: center; } +.seal-circle .co { font-size: 9.5px; letter-spacing: .5px; line-height: 13px; padding: 0 8px; } +.seal-circle .txt { font-size: 16px; font-weight: 700; letter-spacing: 3px; margin: 5px 0; } +.seal-circle .who { font-size: 8.5px; line-height: 12px; } diff --git a/po_app/templates/base.html b/po_app/templates/base.html new file mode 100644 index 0000000..e4924fb --- /dev/null +++ b/po_app/templates/base.html @@ -0,0 +1,36 @@ + + + + +{% block title %}采购订单管理系统{% endblock %} + + + +{% if session.get('user_id') %} +
+ 博特维 · 采购订单管理 + + + {{ session.get('user_name') }}（{{ ROLE_TEXT.get(session.get('role'), '') }}） + 退出 +
+{% endif %} +
+ {% with msgs = get_flashed_messages(with_categories=true) %} + {% for cat, msg in msgs %} +
{{ msg }}
+ {% endfor %} + {% endwith %} + {% block body %}{% endblock %} +
+
{{ COMPANY_NAME }} · 采购订单管理系统 v1.0
+ + diff --git a/po_app/templates/index.html b/po_app/templates/index.html new file mode 100644 index 0000000..fb8f0bf --- /dev/null +++ b/po_app/templates/index.html @@ -0,0 +1,50 @@ +{% extends "base.html" %} +{% block body %} +

工作台

+
+
{{ counts.get('pending_review', 0) }}
待主管审核
+
{{ counts.get('pending_approval', 0) }}
待经理批准
+
{{ counts.get('approved', 0) }}
已批准
+
{{ counts.get('draft', 0) }}
草稿
+
{{ '%.2f'|format(month_total) }}
已批准订单总额（元）
+
+ +{% if todo %} +
+

待我处理（{{ todo|length }}）

+ + + {% for o in todo %} + + + + + + + + + + {% endfor %} +
单号类型供应商金额状态提交时间操作
{{ o.order_no }}{{ o.doc_type_text }}{{ o.supplier_name }}{{ '%.2f'|format(o.total_amount) }}{{ o.status_text }}{{ o.submitted_at.strftime('%m-%d %H:%M') if o.submitted_at else '-' }}去处理
+
+{% endif %} + +
+

我的订单

+ {% if my_orders %} + + + {% for o in my_orders %} + + + + + + + + + {% endfor %} +
单号类型供应商金额状态创建时间
{{ o.order_no }}{{ o.doc_type_text }}{{ o.supplier_name }}{{ '%.2f'|format(o.total_amount) }}{{ o.status_text }}{{ o.created_at.strftime('%Y-%m-%d %H:%M') }}
+ {% else %}

暂无订单，点右上角「新建订单」开始。

{% endif %} +
+{% endblock %} diff --git a/po_app/templates/login.html b/po_app/templates/login.html new file mode 100644 index 0000000..98a0bdc --- /dev/null +++ b/po_app/templates/login.html @@ -0,0 +1,31 @@ + + + + +登录 · 采购订单管理系统 + + + + + + diff --git a/po_app/templates/materials.html b/po_app/templates/materials.html new file mode 100644 index 0000000..0689784 --- /dev/null +++ b/po_app/templates/materials.html @@ -0,0 +1,125 @@ +{% extends "base.html" %} +{% block body %} +

物料库

+
+

新增物料

+
+
+
+
+
+
+
+
+
+
+ +
+
+

物料列表

+ +
+ + + {% for m in rows %} + + + + + + {% endfor %} +
物料代码物料规格单位默认含税单价操作
{{ m.code }}{{ m.spec }}{{ m.unit }}{{ '%.2f'|format(m.price) }} + +
+ +
+ {% if user.role == 'admin' %} +
+ +
+ {% endif %} +
+
+ + + + + + +{% endblock %} diff --git a/po_app/templates/order_edit.html b/po_app/templates/order_edit.html new file mode 100644 index 0000000..ab8f624 --- /dev/null +++ b/po_app/templates/order_edit.html @@ -0,0 +1,201 @@ +{% extends "base.html" %} +{% block body %} +

编辑订单 · {{ o.order_no }}（{{ o.doc_type_text }}）

+ +
+

单据信息

+
+
+
+
+
+
+ +

供方（供应商）

+
+
+ + + +
+
+
+
+
+
+
+
+ +

需方（本公司）

+
+
+
+
+
+
+ + +
+
+ +

物料明细

+ + + + + + + + + + + + + + {% for it in o.items %} + + + + + + + + + + + + + {% endfor %} +
序物料代码物料规格单位数量含税单价金额交期备注删
{{ loop.index }} + +
+ + + +
+ + + 合计：{{ '%.2f'|format(o.total_amount) }} 元 + {{ total_cn }} +
+ +

补充说明

+ + +
+ + 查看订单 + 预览 PDF +
+
+ +
+
+ + +
+
+
+ +
+

从 Excel 导入明细

+
+ 支持 .xlsx / .xls。系统自动识别表头（物料代码、物料规格、单位、数量、含税单价、交期、备注）， + 没有表头则按前 7 列顺序读取。行数超出模板也没关系，PDF 会自动分页。 + 下载导入模板 +
+
+ + + +
+
+ + +{% endblock %} diff --git a/po_app/templates/order_new.html b/po_app/templates/order_new.html new file mode 100644 index 0000000..ff36ae7 --- /dev/null +++ b/po_app/templates/order_new.html @@ -0,0 +1,30 @@ +{% extends "base.html" %} +{% block body %} +

新建订单

+
+
先选单据类型与供应商，系统会带出供应商资料并生成单号，下一步填写物料明细。
+
+
+
+ + +
+
+ + +
+
+
+ + 取消 +
+
+
+{% endblock %} diff --git a/po_app/templates/order_view.html b/po_app/templates/order_view.html new file mode 100644 index 0000000..06c7a3f --- /dev/null +++ b/po_app/templates/order_view.html @@ -0,0 +1,203 @@ +{% extends "base.html" %} +{% block body %} +

{{ o.doc_type_text }} · {{ o.order_no }} + {{ o.status_text }} +

+ +
+ {% if editable %}编辑{% endif %} + 导出 / 预览 PDF + 下载 PDF +
+ +
+ + 返回列表 +
+ +
+

单据信息

+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
供应商{{ o.supplier_name }}需方{{ o.our_company }}
联系人{{ o.supplier_contact }}联系人{{ o.our_contact }}
电话{{ o.supplier_phone }}电话{{ o.our_phone }}
付款方式{{ o.payment_terms }}税点{{ o.tax_rate }}
税号{{ o.supplier_tax_no }}税号{{ o.our_tax_no }}
银行帐号{{ o.supplier_bank_account }}采购单号{{ o.order_no }}
开户银行{{ o.supplier_bank_name }}采购日期{{ o.order_date }}
交货地点{{ o.delivery_place }}
+ +

物料明细

+ + + + {% for it in o.items %} + + + + + + + + + + + + {% else %} + + {% endfor %} +
序号物料代码物料规格单位数量含税单价金额交期备注
{{ loop.index }}{{ it.material_code }}{{ it.material_spec }}{{ it.unit }}{{ it.qty }}{{ '%.2f'|format(it.price) }}{{ '%.2f'|format(it.amount) }}{{ it.delivery_date }}{{ it.remark }}
暂无明细
+
+ 合计（小写）：{{ '%.2f'|format(o.total_amount) }} 元 + 大写：{{ total_cn }} +
+ {% if o.note %}

补充说明：{{ o.note }}

{% endif %} +
+ +
+

审核流程

+ + + + + +
环节操作人时间
制单{{ o.created_by_name }}{{ o.created_at.strftime('%Y-%m-%d %H:%M') if o.created_at else '-' }}
主管审核{{ o.reviewed_by or '—' }}{{ o.reviewed_at.strftime('%Y-%m-%d %H:%M') if o.reviewed_at else '—' }}
经理批准{{ o.approved_by or '—' }}{{ o.approved_at.strftime('%Y-%m-%d %H:%M') if o.approved_at else '—' }}
+ + {% if user.role in ['reviewer','admin'] and o.status == 'pending_review' %} +
+ +
+ + + + +
+
流程：先点「加盖审核章」，确认无误后再点「确认审核通过」。
+
+ + {% endif %} + {% if user.role in ['approver','admin'] and o.status == 'pending_approval' %} +
+ +
+ + + + +
+
流程：先点「加盖批准章」，确认无误后再点「确认批准通过」。
+
+ + {% endif %} + {% if user.role in ['reviewer','approver','admin'] and o.status in ['pending_review','pending_approval'] %} +
+ + +
+ {% endif %} + {% if editable and o.status in ['draft','rejected'] %} +
+ + +
+ {% endif %} + {% if o.status == 'rejected' and o.reject_reason %} +
驳回原因：{{ o.reject_reason }}
+ {% endif %} + +

审核记录

+
    + {% for lg in o.logs %} +
  • {{ lg.created_at.strftime('%Y-%m-%d %H:%M') }} · {{ lg.user_name }} · {{ lg.action }} + {% if lg.comment %}（{{ lg.comment }}）{% endif %}
    • + {% else %} +
  • 暂无记录
    • + {% endfor %} +
+
+{% endblock %} diff --git a/po_app/templates/orders.html b/po_app/templates/orders.html new file mode 100644 index 0000000..8132514 --- /dev/null +++ b/po_app/templates/orders.html @@ -0,0 +1,53 @@ +{% extends "base.html" %} +{% block body %} +

订单列表

+
+
+ + + + + ~ + + + + 导出 Excel 台账 + 新建订单 +
+ + + + + + {% for o in rows %} + + + + + + + + + + + + + {% else %} + + {% endfor %} +
单号类型供应商采购日期金额状态制单审核批准操作
{{ o.order_no }}{{ o.doc_type_text }}{{ o.supplier_name }}{{ o.order_date }}{{ '%.2f'|format(o.total_amount) }}{{ o.status_text }}{{ o.created_by_name }}{{ o.reviewed_by or '-' }}{{ o.approved_by or '-' }} + 查看 + PDF +
没有符合条件的订单
+
+{% endblock %} diff --git a/po_app/templates/stats.html b/po_app/templates/stats.html new file mode 100644 index 0000000..e33a92f --- /dev/null +++ b/po_app/templates/stats.html @@ -0,0 +1,40 @@ +{% extends "base.html" %} +{% block body %} +

统计

+
+

按状态

+ + + {% for st, cnt, amt in by_status %} + + + + + + {% endfor %} +
状态订单数金额合计（元）
{{ status_text.get(st, st) }}{{ cnt }}{{ '%.2f'|format(amt or 0) }}
+
+ +
+

按供应商（仅已批准）

+ + + {% for name, cnt, amt in by_supplier %} + + {% else %}{% endfor %} +
供应商订单数金额合计（元）
{{ name }}{{ cnt }}{{ '%.2f'|format(amt or 0) }}
暂无数据
+
+ +
+

按月份

+ + + {% for m, cnt, amt in by_month %} + + {% endfor %} +
月份订单数金额合计（元）
{{ m }}{{ cnt }}{{ '%.2f'|format(amt or 0) }}
+ +
+{% endblock %} diff --git a/po_app/templates/suppliers.html b/po_app/templates/suppliers.html new file mode 100644 index 0000000..6566dd9 --- /dev/null +++ b/po_app/templates/suppliers.html @@ -0,0 +1,132 @@ +{% extends "base.html" %} +{% block body %} +

供应商管理

+
+

新增供应商

+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ +
+
+

供应商列表

+ +
+ + + {% for s in rows %} + + + + + + + {% endfor %} +
编号名称联系人电话付款方式税号开户银行银行帐号操作
{{ s.code }}{{ s.name }}{{ s.contact }}{{ s.phone }}{{ s.payment_terms }}{{ s.tax_no }}{{ s.bank_name }}{{ s.bank_account }} + +
+ +
+ {% if user.role == 'admin' %} +
+ +
+ {% endif %} +
+
+ + + + + + +{% endblock %} diff --git a/po_app/templates/users.html b/po_app/templates/users.html new file mode 100644 index 0000000..e287740 --- /dev/null +++ b/po_app/templates/users.html @@ -0,0 +1,39 @@ +{% extends "base.html" %} +{% block body %} +

用户与权限

+
+

新增用户

+
+
+
+
+
+ +
+
+
+
+
+
+ +
+

用户列表

+ + + {% for t in rows %} + + + + + + + {% endfor %} +
登录名姓名角色状态创建时间
{{ t.username }}{{ t.name }}{{ roles.get(t.role, t.role) }}{{ '启用' if t.active else '停用' }}{{ t.created_at.strftime('%Y-%m-%d %H:%M') if t.created_at else '-' }}
+
+ 制单员：新建 / 编辑 / 提交；主管审核：审核通过或驳回；经理批准：批准后订单生效并可导出正式 PDF； + 管理员：全部权限，含用户、供应商、物料维护。 +
+
+{% endblock %} diff --git a/po_app/使用说明.md b/po_app/使用说明.md new file mode 100644 index 0000000..26e8207 --- /dev/null +++ b/po_app/使用说明.md @@ -0,0 +1,77 @@ +# 博特维采购订单管理系统 v1.0 + +基于贵司两个 Excel 模板（采购合同模板 / 委外生产订单模板）开发的本地采购订单管理系统： +制单 → 主管审核 → 经理批准 → 导出 PDF。 + +## 一、启动 + +三种方式，任选其一： + +1. **单文件 EXE（推荐分发）**：`dist\博特维采购订单系统_单文件版.exe`（36.7 MB），拷到任意文件夹双击即用，免 Python 环境 +2. **文件夹版 EXE（启动最快）**：`dist\博特维采购订单系统\` 整个文件夹，双击里面的 exe 秒开 +3. **源码运行**：双击 `启动.bat`，或命令行 `python app.py 8080` + +访问地址 http://127.0.0.1:8080；局域网其他电脑把 `127.0.0.1` 换成本机 IP（需放行 8080 端口）。 +EXE 版数据在 exe 同级自动生成的 `instance\po.db`，备份即复制该文件夹。 + +> 源码方式依赖：Flask、SQLAlchemy、reportlab、openpyxl、xlrd（`pip install -r requirements.txt`） + +## 二、初始账号 + +| 账号 | 密码 | 角色 | 权限 | +|---|---|---|---| +| admin | admin123 | 系统管理员 | 全部功能，含用户/供应商/物料维护 | +| maker | 123456 | 制单员 | 新建、编辑、提交订单 | +| reviewer | 123456 | 主管审核 | 审核通过或驳回 | +| approver | 123456 | 经理批准 | 批准后订单生效、可出正式 PDF | + +> 正式使用前请在「用户」页改掉默认密码。 + +## 三、流程 + +``` +制单员新建订单 → 填写抬头与明细 → 提交 + ↓ +主管审核（通过 / 驳回并填原因） + ↓ +经理批准（通过 / 驳回并填原因） + ↓ +状态 = 已批准 → 导出 PDF（无水印，带审核人姓名与时间） +``` + +- 草稿 / 待审核 / 待批准状态导出的 PDF 会带斜向水印（草稿、待审核等），避免误用 +- 被驳回的订单回到制单员名下，可修改后再次提交 +- 每个动作都会写入「审核记录」，可追溯 + +## 四、功能说明 + +| 功能 | 说明 | +|---|---| +| 两种单据 | 采购单 / 委外生产订单共用数据，版式按原模板分别套打 | +| Excel 导入明细 | 编辑页上传 .xls/.xlsx，自动识别表头（代码/规格/单位/数量/单价/交期/备注）；行数不限，PDF 自动分页 | +| 供应商库 | 制单时下拉带入名称、联系人、电话、税号、银行、付款方式 | +| 物料库 | 明细行下拉带入代码、规格、单位、单价 | +| 金额大写 | 合计自动转人民币大写（壹万贰仟叁佰肆拾伍元整） | +| 历史查询 | 按单号、供应商、状态、单据类型、日期区间筛选 | +| 统计 | 按状态 / 供应商 / 月份汇总金额 | +| 台账导出 | 一键导出全部订单为 Excel | + +## 五、PDF 版式 + +完全复刻原 Excel 模板： + +1. 公司名 + 单据名称（采购单 / 委外生产订单） +2. 抬头区：左供方（供应商、联系人、电话、付款方式、税号、银行帐号、开户银行）、右需方（需方、联系人、电话、税点、税号、采购单号、采购日期） +3. 明细表 9 列：序号、物料代码、物料规格、单位、数量、含税单价、金额、交期、备注（列宽取自原模板，两种单据各自比例） +4. 合计：小写 + 人民币大写 + 元整 +5. 备注条款 9 条 + 交货地点（苏州工业园区星汉街 5 号腾飞新苏工业坊 B 幢 5 楼 05-06 单元 胡乃琛 13913172516） +6. 签署区：制单 / 主管审核 / 经理批准（含姓名与时间）+ 双方盖章栏 +7. 页脚：单据类型、单号、页码、打印时间 + +明细超过一页时自动分页并重复表头，合计与条款跟随在最后一页。 + +## 六、数据文件 + +- 数据库：`instance/po.db`（SQLite，整个文件夹复制即可备份） +- 导出的 PDF：`exports/` +- 上传的明细文件：`uploads/` diff --git a/po_app/博特维采购订单系统.spec b/po_app/博特维采购订单系统.spec new file mode 100644 index 0000000..c356bf3 --- /dev/null +++ b/po_app/博特维采购订单系统.spec @@ -0,0 +1,51 @@ +# -*- mode: python ; coding: utf-8 -*- +from PyInstaller.utils.hooks import collect_all + +datas = [('templates', 'templates'), ('static', 'static')] +binaries = [] +hiddenimports = ['sqlalchemy.dialects.sqlite', 'openpyxl', 'xlrd'] +tmp_ret = collect_all('reportlab') +datas += tmp_ret[0]; binaries += tmp_ret[1]; hiddenimports += tmp_ret[2] + + +a = Analysis( + ['app.py'], + pathex=[], + binaries=binaries, + datas=datas, + hiddenimports=hiddenimports, + hookspath=[], + hooksconfig={}, + runtime_hooks=[], + excludes=[], + noarchive=False, + optimize=0, +) +pyz = PYZ(a.pure) + +exe = EXE( + pyz, + a.scripts, + [], + exclude_binaries=True, + name='博特维采购订单系统', + debug=False, + bootloader_ignore_signals=False, + strip=False, + upx=True, + console=True, + disable_windowed_traceback=False, + argv_emulation=False, + target_arch=None, + codesign_identity=None, + entitlements_file=None, +) +coll = COLLECT( + exe, + a.binaries, + a.datas, + strip=False, + upx=True, + upx_exclude=[], + name='博特维采购订单系统', +) diff --git a/po_app/博特维采购订单系统_单文件版.spec b/po_app/博特维采购订单系统_单文件版.spec new file mode 100644 index 0000000..4888810 --- /dev/null +++ b/po_app/博特维采购订单系统_单文件版.spec @@ -0,0 +1,45 @@ +# -*- mode: python ; coding: utf-8 -*- +from PyInstaller.utils.hooks import collect_all + +datas = [('templates', 'templates'), ('static', 'static')] +binaries = [] +hiddenimports = ['sqlalchemy.dialects.sqlite', 'openpyxl', 'xlrd'] +tmp_ret = collect_all('reportlab') +datas += tmp_ret[0]; binaries += tmp_ret[1]; hiddenimports += tmp_ret[2] + + +a = Analysis( + ['app.py'], + pathex=[], + binaries=binaries, + datas=datas, + hiddenimports=hiddenimports, + hookspath=[], + hooksconfig={}, + runtime_hooks=[], + excludes=[], + noarchive=False, + optimize=0, +) +pyz = PYZ(a.pure) + +exe = EXE( + pyz, + a.scripts, + a.binaries, + a.datas, + [], + name='博特维采购订单系统_单文件版', + debug=False, + bootloader_ignore_signals=False, + strip=False, + upx=True, + upx_exclude=[], + runtime_tmpdir=None, + console=True, + disable_windowed_traceback=False, + argv_emulation=False, + target_arch=None, + codesign_identity=None, + entitlements_file=None, +) diff --git a/po_app/启动.bat b/po_app/启动.bat new file mode 100755 index 0000000..c512fd7 --- /dev/null +++ b/po_app/启动.bat @@ -0,0 +1,16 @@ +@echo off +chcp 65001 >nul +cd /d "%~dp0" + +set "PY=%USERPROFILE%\.workbuddy\binaries\python\envs\default\Scripts\python.exe" +if not exist "%PY%" set "PY=python" + +echo ============================================ +echo 博特维采购订单管理系统 +echo 浏览器访问： http://127.0.0.1:8080 +echo 关闭本窗口即停止服务 +echo ============================================ + +start "" http://127.0.0.1:8080 +"%PY%" app.py 8080 +pause diff --git a/tools/dump_xls.py b/tools/dump_xls.py new file mode 100644 index 0000000..5857753 --- /dev/null +++ b/tools/dump_xls.py @@ -0,0 +1,40 @@ +# -*- coding: utf-8 -*- +"""Dump .xls (BIFF) content: sheets, merged cells, all non-empty cell values.""" +import sys +import xlrd + +def dump(path): + print("=" * 100) + print("FILE:", path) + try: + bk = xlrd.open_workbook(path, formatting_info=True) + fi = True + except Exception as e: + print(" [formatting_info unavailable: %s]" % e) + bk = xlrd.open_workbook(path) + fi = False + print("sheets:", bk.sheet_names(), " nsheets:", bk.nsheets) + for sh in bk.sheets(): + print("-" * 100) + print("SHEET: %s (%d rows x %d cols)" % (sh.name, sh.nrows, sh.ncols)) + if fi: + try: + merged = sh.merged_cells + if merged: + print("MERGED:", merged) + except Exception: + pass + for r in range(sh.nrows): + row_vals = [] + for c in range(sh.ncols): + v = sh.cell_value(r, c) + if isinstance(v, float) and v == int(v): + v = int(v) + s = str(v).strip() + if s: + row_vals.append("%s%d=%s" % (chr(65 + c) if c < 26 else "c" + str(c), r + 1, s)) + if row_vals: + print(" | ".join(row_vals)) + +for p in sys.argv[1:]: + dump(p) diff --git a/tools/e2e_test.py b/tools/e2e_test.py new file mode 100644 index 0000000..d654889 --- /dev/null +++ b/tools/e2e_test.py @@ -0,0 +1,99 @@ +# -*- coding: utf-8 -*- +"""端到端冒烟：制单 → 提交 → 主管审核 → 经理批准 → 导出 PDF。""" +import http.cookiejar +import os +import re +import urllib.parse +import urllib.request + +BASE = "http://127.0.0.1:8080" +OUT = r"D:\2026-09-09-task-18\tools\out" +os.makedirs(OUT, exist_ok=True) + + +def new_session(): + cj = http.cookiejar.CookieJar() + return urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj)) + + +def post(op, path, data): + body = urllib.parse.urlencode(data).encode() + req = urllib.request.Request(BASE + path, data=body) + r = op.open(req, timeout=30) + return r.geturl(), r.read().decode("utf-8", "replace") + + +def get(op, path): + r = op.open(BASE + path, timeout=30) + return r.geturl(), r.read() + + +def login(user, pwd): + op = new_session() + url, html = post(op, "/login", {"username": user, "password": pwd}) + assert "/login" not in url, "%s 登录失败" % user + print(" login %s ok" % user) + return op + + +def make_order(op, doc_type, supplier_id, n_items): + url, _ = post(op, "/orders/new", {"doc_type": doc_type, "supplier_id": supplier_id}) + m = re.search(r"/orders/(\d+)", url) + oid = int(m.group(1)) + form = [ + ("order_date", "2026-09-09"), ("tax_rate", "13%"), + ("supplier_name", "苏州精锐五金制品有限公司"), + ("supplier_contact", "周经理"), ("supplier_phone", "13800000001"), + ("payment_terms", "月结30天"), ("supplier_tax_no", "91320500MA1XXXXXXX"), + ("supplier_bank_account", "4562 1088 0012 3456"), + ("supplier_bank_name", "中国银行苏州工业园区支行"), + ("our_company", "苏州博特维智能科技有限公司"), + ("our_contact", "胡乃琛"), ("our_phone", "13913172516"), + ("our_tax_no", "91320594MA1YYYYYYY"), + ("delivery_place", "苏州市工业园区星汉街5号腾飞新苏工业坊B幢5楼05-06单元 胡乃琛 13913172516"), + ("note", "本单为系统测试单，含 & <特殊> 字符校验"), + ] + for i in range(n_items): + form += [ + ("item_code", "M-10%02d" % (i + 1)), + ("item_spec", "测试物料 规格 A&B <%d>" % (i + 1)), + ("item_unit", "件"), ("item_qty", "12.5"), ("item_price", "38.5"), + ("item_date", "2026-09-20"), ("item_remark", "行%d" % (i + 1)), + ] + post(op, "/orders/%d" % oid, form) + return oid + + +def run(tag, doc_type, n_items): + print("[%s] doc=%s items=%d" % (tag, doc_type, n_items)) + maker = login("maker", "123456") + oid = make_order(maker, doc_type, 1, n_items) + print(" 制单 ok, id=%d" % oid) + post(maker, "/orders/%d/submit" % oid, {"comment": "请审核"}) + + rv = login("reviewer", "123456") + post(rv, "/orders/%d/review" % oid, {"comment": "同意"}) + + ap = login("approver", "123456") + post(ap, "/orders/%d/approve" % oid, {"comment": "批准"}) + print(" 审核流程 ok") + + _, data = get(ap, "/orders/%d/pdf" % oid) + path = os.path.join(OUT, "%s.pdf" % tag) + with open(path, "wb") as f: + f.write(data) + print(" PDF -> %s (%d bytes)" % (path, len(data))) + + # 草稿水印版 + oid2 = make_order(maker, doc_type, 1, 2) + _, data2 = get(maker, "/orders/%d/pdf" % oid2) + p2 = os.path.join(OUT, "%s_draft.pdf" % tag) + with open(p2, "wb") as f: + f.write(data2) + print(" 草稿 PDF -> %s (%d bytes)" % (p2, len(data2))) + + +if __name__ == "__main__": + run("purchase", "purchase", 3) + run("outsourcing_30rows", "outsourcing", 30) + print("ALL DONE") diff --git a/tools/git_init_commit.py b/tools/git_init_commit.py new file mode 100644 index 0000000..a94695d --- /dev/null +++ b/tools/git_init_commit.py @@ -0,0 +1,49 @@ +# -*- coding: utf-8 -*- +"""Init local git repo + commit source (respecting .gitignore) using dulwich. +No network/credentials needed. Push happens separately once remote URL+token are known. +""" +import os, sys +from dulwich import porcelain +from dulwich.repo import Repo + +ROOT = r"D:\2026-09-09-task-18" +GITDIR = os.path.join(ROOT, ".git") + +# 1) init (skip if already a repo) +if os.path.exists(GITDIR): + print("repo already exists, reusing") + repo = Repo(GITDIR) +else: + repo = porcelain.init(ROOT) + print("repo initialized at", GITDIR) + +# 2) stage everything; dulwich add honors .gitignore via ignore manager +porcelain.add(repo, [ROOT]) +print("staged files (count):", len(repo.open_index())) + +# 3) show what got staged (verify no db / dist / backups) +staged = list(repo.open_index()) +bad = [p for p in staged if + p.endswith(b".db") or b"\\dist\\" in p or p.startswith(b"dist/") or + b"dist_backup" in p or b"\\instance\\" in p or b"/instance/" in p or + p.startswith(b".workbuddy")] +print("SUSPECT_STAGED:", [p.decode("utf-8","replace") for p in bad]) + +if bad: + print("ABORT: ignored files would be committed; fix .gitignore") + sys.exit(2) + +# 4) commit (with identity) +porcelain.commit(repo, + message=b"init: \xe5\x8d\x9a\xe7\x89\xb9\xe7\xbb\xb4\xe9\x87\x87\xe8\xb4\xad\xe8\xae\xa2\xe5\x8d\x95\xe7\xb3\xbb\xe7\xbb\x9f v1.0 \xe6\xba\x90\xe7\xa0\x81", + author=b"star ", + committer=b"star ") +print("COMMITTED. HEAD:", repo.head().decode()) + +# 5) report tracked file list +tracked = [] +for e in repo.open_index().iteritems(): + tracked.append(e[0].decode("utf-8","replace")) +print("TRACKED_FILES:") +for f in sorted(tracked): + print(" ", f) diff --git a/tools/pdf_preview.py b/tools/pdf_preview.py new file mode 100644 index 0000000..4394a7b --- /dev/null +++ b/tools/pdf_preview.py @@ -0,0 +1,28 @@ +# -*- coding: utf-8 -*- +"""把 PDF 指定页渲染成 PNG，用于版式校验。""" +import os +import sys + +import pymupdf + +OUT = r"D:\2026-09-09-task-18\tools\out" + + +def render(pdf, pages=(0,), dpi=110, prefix=None): + doc = pymupdf.open(pdf) + prefix = prefix or os.path.splitext(os.path.basename(pdf))[0] + for p in pages: + if p >= len(doc): + continue + page = doc[p] + pix = page.get_pixmap(dpi=dpi) + path = os.path.join(OUT, "%s_p%d.png" % (prefix, p + 1)) + pix.save(path) + print(path) + print("pages:", len(doc)) + + +if __name__ == "__main__": + name = sys.argv[1] + pages = [int(x) for x in sys.argv[2].split(",")] if len(sys.argv) > 2 else [0] + render(os.path.join(OUT, name), pages) diff --git a/tools/run_check.py b/tools/run_check.py new file mode 100644 index 0000000..0f36e37 --- /dev/null +++ b/tools/run_check.py @@ -0,0 +1,13 @@ +# -*- coding: utf-8 -*- +import subprocess +import sys + +PY = r"C:\Users\wps18\.workbuddy\binaries\python\envs\default\Scripts\python.exe" +CWD = r"D:\2026-09-09-task-18\po_app" + +p = subprocess.run([PY, "-c", "import app; print('IMPORT OK')"], cwd=CWD, + capture_output=True, text=True, encoding="utf-8", errors="replace", + timeout=60) +print("RC", p.returncode) +print("STDOUT:", p.stdout[-4000:]) +print("STDERR:", p.stderr[-4000:]) diff --git a/tools/test_approve_stamp.py b/tools/test_approve_stamp.py new file mode 100644 index 0000000..30f8af1 --- /dev/null +++ b/tools/test_approve_stamp.py @@ -0,0 +1,100 @@ +# -*- coding: utf-8 -*- +"""模拟：经理批准也需先加章；批准通过后 PDF 上审核章与批准章并排。 + +用法：python test_approve_stamp.py 8099 +""" +import os +import re +import sqlite3 +import sys + +import requests + +BASE = "http://127.0.0.1:%s" % (sys.argv[1] if len(sys.argv) > 1 else "8099") +HERE = os.path.dirname(os.path.abspath(__file__)) +OUT = os.path.join(HERE, "out") +os.makedirs(OUT, exist_ok=True) +DB = os.path.join(os.path.dirname(HERE), "po_app", "instance", "po.db") +FAILS = [] + + +def check(label, cond, extra=""): + print((" OK " if cond else " FAIL ") + label + (" " + str(extra) if extra else "")) + if not cond: + FAILS.append(label) + + +def sess(u, p): + s = requests.Session() + s.post(BASE + "/login", data={"username": u, "password": p}) + return s + + +def flash_of(h): + return " | ".join(re.sub(r"<[^>]+>", "", m.group(1)).strip() + for m in re.finditer(r'class="flash[^"]*"[^>]*>(.*?)', h, re.S)) + + +def dbq(sql, a=()): + con = sqlite3.connect(DB) + cur = con.cursor() + cur.execute(sql, a) + r = cur.fetchone() + con.close() + return r + + +def main(): + admin = sess("admin", "admin123") + r = admin.post(BASE + "/orders/new", data={"doc_type": "purchase"}) + oid = int(re.search(r"/orders/(\d+)/edit", r.url).group(1)) + admin.post(BASE + "/orders/%d" % oid, data={ + "supplier_name": u"测试供应商", + "item_code": ["M-1001"], "item_spec": [u"测试物料"], "item_unit": [u"件"], + "item_qty": ["10"], "item_price": ["12.5"], "item_date": [""], "item_remark": [""], + }) + admin.post(BASE + "/orders/%d/submit" % oid, data={"comment": u"模拟"}) + + print("\n[1] 主管审核（加章）") + rv = sess("reviewer", "123456") + rv.post(BASE + "/orders/%d/review" % oid, data={"comment": u"ok", "stamped": "1"}) + check("审核通过", dbq("SELECT status FROM orders WHERE id=?", (oid,))[0] == "pending_approval") + + print("\n[2] 经理批准：未加章应被拒绝") + ap = sess("approver", "123456") + h = ap.post(BASE + "/orders/%d/approve" % oid, data={"comment": u"直接批", "stamped": "0"}).text + check("未加章 → 提示", u"请先加盖批准章" in flash_of(h), flash_of(h)) + check("状态未变", dbq("SELECT status FROM orders WHERE id=?", (oid,))[0] == "pending_approval") + + print("\n[3] 加章后批准") + ap.post(BASE + "/orders/%d/approve" % oid, data={"comment": u"同意", "stamped": "1"}) + row = dbq("SELECT status, approved_by FROM orders WHERE id=?", (oid,)) + check("已批准", row[0] == "approved", row) + + print("\n[4] 导出 PDF（应含两枚章）") + b = admin.get(BASE + "/orders/%d/pdf" % oid).content + p = os.path.join(OUT, "seal_both.pdf") + open(p, "wb").write(b) + check("PDF 生成", len(b) > 10000, "%d bytes" % len(b)) + try: + import pymupdf + doc = pymupdf.open(p) + doc[-1].get_pixmap(dpi=130).save(os.path.join(OUT, "seal_both.png")) + print(" 预览:", os.path.join(OUT, "seal_both.png"), "%d 页" % doc.page_count) + doc.close() + except Exception as e: + print(" 渲染失败:", e) + + print("\n[5] 清理") + con = sqlite3.connect(DB) + cur = con.cursor() + cur.execute("DELETE FROM order_items WHERE order_id=?", (oid,)) + cur.execute("DELETE FROM orders WHERE id=?", (oid,)) + con.commit() + con.close() + print(" 已删除订单 %d" % oid) + print("\n" + ("全部通过" if not FAILS else "失败: " + "; ".join(FAILS))) + + +if __name__ == "__main__": + main() diff --git a/tools/test_batch_import.py b/tools/test_batch_import.py new file mode 100644 index 0000000..90aa1bc --- /dev/null +++ b/tools/test_batch_import.py @@ -0,0 +1,208 @@ +# -*- coding: utf-8 -*- +"""批量导入 + 需方预置 + PO 号规则 E2E 测试。 + +用法：先启动服务（python app.py 8099），再运行本脚本。 +覆盖：物料/供应商导入的 新增 / 更新 / 跳过 / 失败 四条路径， + 模板下载，PO 号 4 位规则，以及旧格式单号的兼容解析。 +""" +import datetime +import os +import re +import sqlite3 +import sys + +import openpyxl +import requests + +PORT = sys.argv[1] if len(sys.argv) > 1 else "8099" +BASE = "http://127.0.0.1:%s" % PORT +HERE = os.path.dirname(os.path.abspath(__file__)) +OUT = os.path.join(HERE, "out") +os.makedirs(OUT, exist_ok=True) +DB = os.path.join(os.path.dirname(HERE), "po_app", "instance", "po.db") + +s = requests.Session() +FAILS = [] + + +def check(label, cond, extra=""): + print((" OK " if cond else " FAIL ") + label + (" " + str(extra) if extra else "")) + if not cond: + FAILS.append(label) + + +def login(): + r = s.post(BASE + "/login", + data={"username": "admin", "password": "admin123"}, + allow_redirects=False) + return r.status_code in (200, 302) + + +def make_material_xlsx(path): + wb = openpyxl.Workbook() + ws = wb.active + ws.append([u"物料代码", u"物料规格", u"单位", u"默认含税单价"]) + ws.append(["M-9001", u"测试新增物料A", u"件", 12.5]) # 新增 + ws.append(["M-9002", u"测试新增物料B", u"台", 88.0]) # 新增 + ws.append(["M-1001", u"铝合金支架(已存在-应被更新)", u"件", 99.9]) # 已存在 + ws.append(["", u"缺代码应失败", u"件", 1.0]) # 失败 + wb.save(path) + + +def make_supplier_xlsx(path): + wb = openpyxl.Workbook() + ws = wb.active + ws.append([u"编号", u"名称", u"联系人", u"电话", u"付款方式", u"税号", + u"开户银行", u"银行帐号"]) + ws.append(["S901", u"测试新增供应商A", u"王经理", "13700000001", + u"月结30天", "91320500TEST001", u"工商银行苏州分行", "1111 2222 3333"]) + ws.append(["S001", u"苏州精锐五金制品有限公司", u"周经理(已更新)", "13800000001", + u"月结60天", "91320500MA1XXXXXXX", u"中国银行苏州工业园区支行", + "4562 1088 0012 3456"]) # 已存在 + ws.append(["S902", u"", u"缺名称", "13900000009", u"", u"", u"", u""]) # 失败 + wb.save(path) + + +def post_import(url, path, mode): + with open(path, "rb") as f: + return s.post(url, + files={"file": (os.path.basename(path), f, + "application/vnd.openxmlformats-officedocument." + "spreadsheetml.sheet")}, + data={"mode": mode}) + + +def flash_texts(html): + out = [] + for m in re.finditer(r'class="flash[^"]*"[^>]*>(.*?)', html, re.S): + out.append(re.sub(r"<[^>]+>", "", m.group(1)).strip()) + return out + + +def main(): + print("登录:", "OK" if login() else "FAIL") + # 预清理：让脚本可重复运行（上次残留的测试数据不影响本次断言） + try: + today0 = datetime.datetime.now().strftime("%Y%m%d") + con = sqlite3.connect(DB) + cur = con.cursor() + cur.execute("UPDATE materials SET spec=?, unit=?, price=? WHERE code=?", + (u"铝合金支架 6061-T6 200×80×10mm", u"件", 38.5, "M-1001")) + cur.execute("UPDATE suppliers SET contact=?, payment_terms=? WHERE code=?", + (u"周经理", u"月结30天", "S001")) + cur.execute("DELETE FROM materials WHERE code IN ('M-9001','M-9002')") + cur.execute("DELETE FROM suppliers WHERE code IN ('S901','S902')") + for p in ("PO", "WO"): + cur.execute("DELETE FROM order_items WHERE order_id IN " + "(SELECT id FROM orders WHERE order_no LIKE ?)", + (p + today0 + "%",)) + cur.execute("DELETE FROM orders WHERE order_no LIKE ?", (p + today0 + "%",)) + con.commit() + con.close() + except Exception as e: + print("预清理失败:", e) + + mp = os.path.join(OUT, "materials_test.xlsx") + sp = os.path.join(OUT, "suppliers_test.xlsx") + make_material_xlsx(mp) + make_supplier_xlsx(sp) + + print("\n[1] 物料导入 update 模式") + r = post_import(BASE + "/materials/import_excel", mp, "update") + t = " | ".join(flash_texts(r.text)) + print(" flash:", t) + check("新增2 / 更新1 / 失败1", u"新增 2" in t and u"更新 1" in t and u"失败 1" in t) + + print("\n[2] 供应商导入 update 模式") + r = post_import(BASE + "/suppliers/import_excel", sp, "update") + t = " | ".join(flash_texts(r.text)) + print(" flash:", t) + check("新增1 / 更新1 / 失败1", u"新增 1" in t and u"更新 1" in t and u"失败 1" in t) + + print("\n[3] 再导一次，skip 模式（应全部跳过）") + r = post_import(BASE + "/materials/import_excel", mp, "skip") + t = " | ".join(flash_texts(r.text)) + print(" 物料 flash:", t) + check("物料跳过3（M-9001/M-9002/M-1001 此时都已存在）", u"跳过 3" in t) + r = post_import(BASE + "/suppliers/import_excel", sp, "skip") + t = " | ".join(flash_texts(r.text)) + print(" 供应商 flash:", t) + check("供应商跳过2（S901/S001 此时都已存在）", u"跳过 2" in t) + + print("\n[4] 列表核对") + html = s.get(BASE + "/materials").text + check("M-9001 已新增", "M-9001" in html) + check("M-1001 单价已更新为 99.90", "99.90" in html) + html = s.get(BASE + "/suppliers").text + check("S901 已新增", "S901" in html) + check("联系人已更新为 周经理(已更新)", u"周经理(已更新)" in html) + + print("\n[5] 模板下载") + for u in ("/materials/import_template.xlsx", "/suppliers/import_template.xlsx"): + rr = s.get(BASE + u) + check(u, rr.status_code == 200 and len(rr.content) > 3000, + "%s bytes" % len(rr.content)) + + print("\n[6] 新建采购单：PO 号规则 + 需方预置") + created = [] + r = s.post(BASE + "/orders/new", data={"doc_type": "purchase"}) + m = re.search(r"PO\d{12}", r.text) + no = m.group(0) if m else "" + created.append(no) + print(" 单号:", no) + today = datetime.datetime.now().strftime("%Y%m%d") + check("格式为 PO+年月日+4位", bool(re.match(r"^PO\d{8}\d{4}$", no))) + check("日期为今天 " + today, no.startswith("PO" + today)) + for k, v in (("our_contact", u"胡乃琛"), ("our_phone", u"13913172516"), + ("our_tax_no", u"91320507MA1WW8D15D"), ("tax_rate", u"13%")): + check("预置 %s = %s" % (k, v), v in r.text) + + print("\n[7] 旧格式单号兼容（插入带横杠的 PO%s-0007）" % today) + legacy = "PO%s-0007" % today + try: + con = sqlite3.connect(DB) + con.execute("INSERT INTO orders (order_no, doc_type, status, order_date) " + "VALUES (?,?,?,?)", + (legacy, "purchase", "draft", + datetime.datetime.now().strftime("%Y-%m-%d"))) + con.commit() + con.close() + r = s.post(BASE + "/orders/new", data={"doc_type": "purchase"}) + m2 = re.search(r"PO\d{12}", r.text) + got = m2.group(0) if m2 else "" + print(" 新单号:", got, " 期望: PO%s0008" % today) + check("旧格式未导致序号重置", got == "PO%s0008" % today) + created.extend([got, legacy]) + except Exception as e: + print(" 跳过，原因:", e) + + print("\n[8] 清理测试数据（还原被更新的既有记录）") + try: + con = sqlite3.connect(DB) + cur = con.cursor() + cur.execute("UPDATE materials SET spec=?, unit=?, price=? WHERE code=?", + (u"铝合金支架 6061-T6 200×80×10mm", u"件", 38.5, "M-1001")) + cur.execute("UPDATE suppliers SET contact=?, payment_terms=? WHERE code=?", + (u"周经理", u"月结30天", "S001")) + cur.execute("DELETE FROM materials WHERE code IN ('M-9001','M-9002')") + cur.execute("DELETE FROM suppliers WHERE code IN ('S901','S902')") + for no_ in [x for x in created if x]: + cur.execute("DELETE FROM order_items WHERE order_id IN " + "(SELECT id FROM orders WHERE order_no=?)", (no_,)) + cur.execute("DELETE FROM orders WHERE order_no=?", (no_,)) + con.commit() + cur.execute("SELECT COUNT(*) FROM materials WHERE code LIKE 'M-900%'") + n1 = cur.fetchone()[0] + cur.execute("SELECT COUNT(*) FROM suppliers WHERE code IN ('S901','S902')") + n2 = cur.fetchone()[0] + con.close() + check("测试物料已清除", n1 == 0) + check("测试供应商已清除", n2 == 0) + except Exception as e: + print(" 清理失败:", e) + + print("\n" + ("全部通过" if not FAILS else "失败项: " + "; ".join(FAILS))) + + +if __name__ == "__main__": + main() diff --git a/tools/test_edit_row.py b/tools/test_edit_row.py new file mode 100644 index 0000000..220cb0a --- /dev/null +++ b/tools/test_edit_row.py @@ -0,0 +1,119 @@ +# -*- coding: utf-8 -*- +"""编辑功能 + 明细行模板 冒烟测试。 + +覆盖： +1) 新建订单（0 条明细，正是"点添加一行没反应"的场景）页面必须带 rowTpl 模板行 +2) 物料列表编辑：POST /materials 带 id 能改库 +3) 供应商列表编辑：POST /suppliers 带 id 能改库 + +用法：先启动服务（python app.py 8099），再 python test_edit_row.py 8099 +""" +import os +import re +import sqlite3 +import sys + +import requests + +PORT = sys.argv[1] if len(sys.argv) > 1 else "8099" +BASE = "http://127.0.0.1:%s" % PORT +HERE = os.path.dirname(os.path.abspath(__file__)) +DB = os.path.join(os.path.dirname(HERE), "po_app", "instance", "po.db") + +s = requests.Session() +FAILS = [] + + +def check(label, cond, extra=""): + print((" OK " if cond else " FAIL ") + label + (" " + str(extra) if extra else "")) + if not cond: + FAILS.append(label) + + +def main(): + r = s.post(BASE + "/login", data={"username": "admin", "password": "admin123"}) + check("登录", r.status_code == 200, r.status_code) + + print("\n[1] 新建订单（0 明细）—— 明细行模板") + r = s.post(BASE + "/orders/new", data={"doc_type": "purchase"}) + html = r.text + no = "" + m = re.search(r"PO\d{12}", html) + if m: + no = m.group(0) + check("新建订单页含 rowTpl 模板", 'id="rowTpl"' in html) + tpl = re.search(r'', html, re.S) + check("模板内含 tr 行", bool(tpl) and "" in tpl.group(1)) + check("模板含 item_code 输入框", bool(tpl) and 'name="item_code"' in tpl.group(1)) + check("模板含 matpick 下拉", bool(tpl) and "matpick" in tpl.group(1)) + n_code = html.count('name="item_code"') + check("当前确实 0 条明细（模板是唯一来源）", n_code == 1, + "item_code 出现 %d 次" % n_code) + + print("\n[2] 物料列表编辑") + con = sqlite3.connect(DB) + cur = con.cursor() + cur.execute("SELECT id FROM materials WHERE code='M-1002'") + row = cur.fetchone() + con.close() + if not row: + check("找到物料 M-1002", False) + else: + mid = row[0] + r = s.post(BASE + "/materials", + data={"id": mid, "code": "M-1002", "spec": u"步进电机(已编辑)", + "unit": u"台", "price": "299.5"}) + check("物料编辑请求成功", r.status_code == 200, r.status_code) + con = sqlite3.connect(DB) + cur = con.cursor() + cur.execute("SELECT spec, unit, price FROM materials WHERE id=?", (mid,)) + got = cur.fetchone() + check("物料已更新", got == (u"步进电机(已编辑)", u"台", 299.5), got) + cur.execute("UPDATE materials SET spec=?, price=? WHERE id=?", + (u"步进电机 57BYG250 2相", 265.0, mid)) + con.commit() + con.close() + print(" 已还原 M-1002") + + print("\n[3] 供应商列表编辑") + con = sqlite3.connect(DB) + cur = con.cursor() + cur.execute("SELECT id FROM suppliers WHERE code='S002'") + row = cur.fetchone() + con.close() + if not row: + check("找到供应商 S002", False) + else: + sid = row[0] + r = s.post(BASE + "/suppliers", + data={"id": sid, "code": "S002", "name": u"昆山宏远电子科技有限公司", + "contact": u"陈主管(已编辑)", "phone": "13900000002", + "payment_terms": u"货到付款", "tax_no": "91320583MA2XXXXXXX", + "bank_name": u"建设银行昆山分行", + "bank_account": "3220 1988 2233 4455"}) + check("供应商编辑请求成功", r.status_code == 200, r.status_code) + con = sqlite3.connect(DB) + cur = con.cursor() + cur.execute("SELECT contact, payment_terms FROM suppliers WHERE id=?", (sid,)) + got = cur.fetchone() + check("供应商已更新", got == (u"陈主管(已编辑)", u"货到付款"), got) + cur.execute("UPDATE suppliers SET contact=? WHERE id=?", (u"陈主管", sid)) + con.commit() + con.close() + print(" 已还原 S002") + + print("\n[4] 清理测试订单 " + no) + if no: + con = sqlite3.connect(DB) + cur = con.cursor() + cur.execute("DELETE FROM order_items WHERE order_id IN " + "(SELECT id FROM orders WHERE order_no=?)", (no,)) + cur.execute("DELETE FROM orders WHERE order_no=?", (no,)) + con.commit() + con.close() + + print("\n" + ("全部通过" if not FAILS else "失败: " + "; ".join(FAILS))) + + +if __name__ == "__main__": + main() diff --git a/tools/test_import.py b/tools/test_import.py new file mode 100644 index 0000000..75d2f3b --- /dev/null +++ b/tools/test_import.py @@ -0,0 +1,73 @@ +# -*- coding: utf-8 -*- +"""测试 Excel 明细导入接口（multipart 上传）。""" +import io +import os +import re +import urllib.parse +import urllib.request + +import openpyxl + +BASE = "http://127.0.0.1:8080" +OUT = r"D:\2026-09-09-task-18\tools\out" +os.makedirs(OUT, exist_ok=True) + + +def new_session(): + import http.cookiejar + cj = http.cookiejar.CookieJar() + return urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj)) + + +def post_form(op, path, data): + body = urllib.parse.urlencode(data).encode() + r = op.open(urllib.request.Request(BASE + path, data=body), timeout=30) + return r.geturl(), r.read() + + +def multipart_upload(op, path, field, filename, content, extra=None): + boundary = "----PoTestBoundary1234" + buf = io.BytesIO() + for k, v in (extra or {}).items(): + buf.write(("--%s\r\nContent-Disposition: form-data; name=\"%s\"\r\n\r\n%s\r\n" + % (boundary, k, v)).encode("utf-8")) + buf.write(("--%s\r\nContent-Disposition: form-data; name=\"%s\"; filename=\"%s\"\r\n" + "Content-Type: application/octet-stream\r\n\r\n" % (boundary, field, filename)) + .encode("utf-8")) + buf.write(content) + buf.write(("\r\n--%s--\r\n" % boundary).encode()) + req = urllib.request.Request(BASE + path, data=buf.getvalue(), method="POST") + req.add_header("Content-Type", "multipart/form-data; boundary=%s" % boundary) + r = op.open(req, timeout=30) + return r.geturl(), r.read().decode("utf-8", "replace") + + +def main(): + op = new_session() + post_form(op, "/login", {"username": "maker", "password": "123456"}) + + wb = openpyxl.Workbook() + ws = wb.active + ws.append([u"物料代码", u"物料规格", u"单位", u"数量", u"含税单价", u"交期", u"备注"]) + ws.append([u"X-001", u"导入测试件 & <一>", u"件", 10, 99.9, "2026-10-01", u"首件"]) + ws.append([u"X-002", u"导入测试件 二", u"套", 3, 128.5, "2026-10-05", u""]) + buf = io.BytesIO() + wb.save(buf) + xlsx = buf.getvalue() + + url, _ = post_form(op, "/orders/new", {"doc_type": "purchase", "supplier_id": ""}) + oid = int(re.search(r"/orders/(\d+)", url).group(1)) + print("order id", oid) + + url, html = multipart_upload(op, "/orders/%d/import_excel" % oid, "file", + "test.xlsx", xlsx, {"mode": "replace"}) + ok = "已导入 2 行明细" in html + print("import rows ok:", ok) + ok2 = "99.90" in html or "99.9" in html + print("price present:", ok2) + assert ok and ok2 + print("IMPORT TEST PASSED") + + +if __name__ == "__main__": + main() diff --git a/tools/test_pages.py b/tools/test_pages.py new file mode 100644 index 0000000..d2facc2 --- /dev/null +++ b/tools/test_pages.py @@ -0,0 +1,28 @@ +# -*- coding: utf-8 -*- +"""快速检查统计页与台账导出。""" +import http.cookiejar +import urllib.parse +import urllib.request + +BASE = "http://127.0.0.1:8080" + + +def main(): + cj = http.cookiejar.CookieJar() + op = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj)) + op.open(urllib.request.Request( + BASE + "/login", + data=urllib.parse.urlencode({"username": "admin", "password": "admin123"}).encode()), + timeout=30) + for path, expect in [("/stats", b"200"), ("/export/ledger", b"xlsx"), + ("/orders", b"200"), ("/suppliers", b"200"), + ("/materials", b"200"), ("/users", b"200"), ("/", b"200")]: + r = op.open(BASE + path, timeout=30) + data = r.read() + print(path, r.status, len(data)) + assert r.status == 200, path + print("PAGES OK") + + +if __name__ == "__main__": + main() diff --git a/tools/test_stamp_toggle.py b/tools/test_stamp_toggle.py new file mode 100644 index 0000000..67b2d16 --- /dev/null +++ b/tools/test_stamp_toggle.py @@ -0,0 +1,156 @@ +# -*- coding: utf-8 -*- +"""模拟完整流程：停用/启用/删除 + 主管审核加章，并导出带章 PDF 渲染成 PNG。 + +用法：先启动服务（python app.py 8099），再 python test_stamp_toggle.py 8099 +""" +import os +import re +import sqlite3 +import sys + +import requests + +PORT = sys.argv[1] if len(sys.argv) > 1 else "8099" +BASE = "http://127.0.0.1:%s" % PORT +HERE = os.path.dirname(os.path.abspath(__file__)) +OUT = os.path.join(HERE, "out") +os.makedirs(OUT, exist_ok=True) +DB = os.path.join(os.path.dirname(HERE), "po_app", "instance", "po.db") + +FAILS = [] + + +def check(label, cond, extra=""): + print((" OK " if cond else " FAIL ") + label + (" " + str(extra) if extra else "")) + if not cond: + FAILS.append(label) + + +def new_sess(u, p): + s = requests.Session() + r = s.post(BASE + "/login", data={"username": u, "password": p}) + assert r.status_code == 200, r.status_code + return s + + +def flash_of(html): + return " | ".join(re.sub(r"<[^>]+>", "", m.group(1)).strip() + for m in re.finditer(r'class="flash[^"]*"[^>]*>(.*?)', + html, re.S)) + + +def row_html(html, key): + """取出包含指定关键字的那个 ...。""" + for m in re.finditer(r"]*>(.*?)", html, re.S): + if key in m.group(1): + return m.group(0) + return "" + + +def dbq(sql, args=()): + con = sqlite3.connect(DB) + cur = con.cursor() + cur.execute(sql, args) + r = cur.fetchone() + con.close() + return r + + +def main(): + admin = new_sess("admin", "admin123") + + print("\n[1] 供应商：停用 → 整行置灰 → 启用 → 彻底删除") + admin.post(BASE + "/suppliers", + data={"code": "S-T1", "name": u"测试供应商T1", "contact": u"测试联系人"}) + sid = dbq("SELECT id FROM suppliers WHERE code='S-T1'")[0] + h = admin.get(BASE + "/suppliers").text + check("S-T1 已创建", "S-T1" in h) + check("初始为启用态（未置灰）", "row-off" not in row_html(h, "S-T1")) + admin.post(BASE + "/suppliers/%d/toggle" % sid) + h = admin.get(BASE + "/suppliers").text + check("停用后整行置灰", "row-off" in row_html(h, "S-T1")) + check("停用后按钮变为『启用』", u"启用" in row_html(h, "S-T1")) + check("停用后仍可编辑（保留入口）", "openEdit" in row_html(h, "S-T1")) + admin.post(BASE + "/suppliers/%d/toggle" % sid) + h = admin.get(BASE + "/suppliers").text + check("启用后恢复（不再置灰）", "row-off" not in row_html(h, "S-T1")) + admin.post(BASE + "/suppliers/%d/delete" % sid) + h = admin.get(BASE + "/suppliers").text + check("删除后从列表消失", "S-T1" not in h) + + print("\n[2] 物料：停用 → 整行置灰 → 启用 → 彻底删除") + admin.post(BASE + "/materials", + data={"code": "M-T1", "spec": u"测试物料T1", "unit": u"件", "price": "9.9"}) + mid = dbq("SELECT id FROM materials WHERE code='M-T1'")[0] + h = admin.get(BASE + "/materials").text + check("M-T1 已创建", "M-T1" in h) + check("初始为启用态（未置灰）", "row-off" not in row_html(h, "M-T1")) + admin.post(BASE + "/materials/%d/toggle" % mid) + h = admin.get(BASE + "/materials").text + check("停用后整行置灰", "row-off" in row_html(h, "M-T1")) + admin.post(BASE + "/materials/%d/toggle" % mid) + h = admin.get(BASE + "/materials").text + check("启用后恢复", "row-off" not in row_html(h, "M-T1")) + admin.post(BASE + "/materials/%d/delete" % mid) + h = admin.get(BASE + "/materials").text + check("删除后从列表消失", "M-T1" not in h) + + print("\n[3] 主管审核：确认前必须先加章") + r = admin.post(BASE + "/orders/new", data={"doc_type": "purchase"}) + m = re.search(r"/orders/(\d+)/edit", r.url) + oid = int(m.group(1)) + admin.post(BASE + "/orders/%d" % oid, data={ + "supplier_name": u"测试供应商", + "item_code": ["M-1001"], "item_spec": [u"测试物料"], + "item_unit": [u"件"], "item_qty": ["10"], "item_price": ["12.5"], + "item_date": [""], "item_remark": [""], + }) + admin.post(BASE + "/orders/%d/submit" % oid, data={"comment": u"模拟提交"}) + st = dbq("SELECT status FROM orders WHERE id=?", (oid,))[0] + check("订单已提交，状态待审核", st == "pending_review", st) + + rv = new_sess("reviewer", "123456") + r = rv.post(BASE + "/orders/%d/review" % oid, + data={"comment": u"直接点确认试试", "stamped": "0"}) + check("未加章 → 被拒绝并提示", u"请先加盖审核章" in flash_of(r.text), + flash_of(r.text)) + st = dbq("SELECT status FROM orders WHERE id=?", (oid,))[0] + check("未加章时状态不变", st == "pending_review", st) + + r = rv.post(BASE + "/orders/%d/review" % oid, + data={"comment": u"已核对无误", "stamped": "1"}) + row = dbq("SELECT status, reviewed_by FROM orders WHERE id=?", (oid,)) + check("加章后审核通过", row[0] == "pending_approval", row) + check("记录了审核人", bool(row[1]), row[1]) + + print("\n[4] 导出带章 PDF 并渲染预览") + pdf_bytes = admin.get(BASE + "/orders/%d/pdf" % oid).content + pdf_path = os.path.join(OUT, "seal_review.pdf") + with open(pdf_path, "wb") as f: + f.write(pdf_bytes) + check("PDF 已生成", len(pdf_bytes) > 10000, "%d bytes" % len(pdf_bytes)) + try: + import fitz + doc = fitz.open(pdf_path) + pix = doc[-1].get_pixmap(dpi=130) + png_path = os.path.join(OUT, "seal_review.png") + pix.save(png_path) + print(" 渲染预览:", png_path, "共 %d 页" % doc.page_count) + doc.close() + except Exception as e: + print(" 渲染预览失败:", e) + + print("\n[5] 清理测试订单") + con = sqlite3.connect(DB) + cur = con.cursor() + cur.execute("DELETE FROM order_items WHERE order_id=?", (oid,)) + cur.execute("DELETE FROM orders WHERE id=?", (oid,)) + con.commit() + con.close() + print(" 已删除订单 %d" % oid) + + print("\n" + ("模拟全部通过" if not FAILS else "失败: " + "; ".join(FAILS))) + + +if __name__ == "__main__": + main() diff --git a/tools/verify_exe.py b/tools/verify_exe.py new file mode 100644 index 0000000..4e0204f --- /dev/null +++ b/tools/verify_exe.py @@ -0,0 +1,82 @@ +# -*- coding: utf-8 -*- +"""打包后 EXE 冒烟测试。 + +确认这些确实进了包：批量导入入口与模板、列表编辑入口、弹窗样式， +以及「新建订单明细为空时仍能添加行」所依赖的 rowTpl 模板。 + +用法：python verify_exe.py 8080 [数据库路径] + 不传数据库路径时默认 po_app/dist/instance/po.db +""" +import os +import re +import sqlite3 +import sys + +import requests + +BASE = "http://127.0.0.1:%s" % (sys.argv[1] if len(sys.argv) > 1 else "8080") +HERE = os.path.dirname(os.path.abspath(__file__)) +DB = sys.argv[2] if len(sys.argv) > 2 else os.path.join( + os.path.dirname(HERE), "po_app", "dist", "instance", "po.db") + +s = requests.Session() +FAILS = [] + + +def check(label, cond, extra=""): + print((" OK " if cond else " FAIL ") + label + (" " + str(extra) if extra else "")) + if not cond: + FAILS.append(label) + + +def main(): + r = s.post(BASE + "/login", data={"username": "admin", "password": "admin123"}) + check("登录", r.status_code == 200, r.status_code) + + # 注意：url_for() 会被 Jinja 渲染成实际路径，页面里不会出现函数名 + for path, marks in ( + ("/materials", [u"批量导入 Excel", "matImpModal", "/materials/import_excel", + "/materials/import_template.xlsx", "openEdit", "editModal", + "/toggle", "/delete", u"停用", u"删除"]), + ("/suppliers", [u"批量导入 Excel", "supImpModal", "/suppliers/import_excel", + "/suppliers/import_template.xlsx", "openEdit", "editModal", + "/toggle", "/delete", u"停用", u"删除"]), + ): + h = s.get(BASE + path).text + missing = [m for m in marks if m not in h] + check("%s 含导入/编辑/停用删除/置灰" % path, not missing, missing) + + css = s.get(BASE + "/static/style.css").text + check("style.css 含弹窗样式", "modal-mask" in css and "modal-box" in css) + + for u in ("/materials/import_template.xlsx", "/suppliers/import_template.xlsx"): + r = s.get(BASE + u) + check("模板下载 %s" % u, r.status_code == 200 and len(r.content) > 3000, + "%s bytes" % len(r.content)) + + # 新建订单：明细为空时必须仍能添加行（依赖 rowTpl 模板存在） + r = s.post(BASE + "/orders/new", data={"doc_type": "purchase"}) + h = r.text + no = "" + m = re.search(r"PO\d{12}", h) + if m: + no = m.group(0) + check("新建订单页含 rowTpl 明细行模板", 'id="rowTpl"' in h) + tpl = re.search(r'', h, re.S) + check("rowTpl 内含完整空行", bool(tpl) and 'name="item_code"' in tpl.group(1)) + + if no and os.path.exists(DB): + con = sqlite3.connect(DB) + cur = con.cursor() + cur.execute("DELETE FROM order_items WHERE order_id IN " + "(SELECT id FROM orders WHERE order_no=?)", (no,)) + cur.execute("DELETE FROM orders WHERE order_no=?", (no,)) + con.commit() + con.close() + print(" 已清理测试订单 %s" % no) + + print("\n" + ("打包版冒烟全部通过" if not FAILS else "失败: " + "; ".join(FAILS))) + + +if __name__ == "__main__": + main() diff --git a/tools/xls_layout.py b/tools/xls_layout.py new file mode 100644 index 0000000..96cf8c1 --- /dev/null +++ b/tools/xls_layout.py @@ -0,0 +1,30 @@ +# -*- coding: utf-8 -*- +"""Dump .xls layout: row heights, col widths, print setup, print area, formulas.""" +import sys +import xlrd + +def dump(path): + print("=" * 90) + print("FILE:", path) + bk = xlrd.open_workbook(path, formatting_info=True) + try: + names = {k: (v.name, getattr(v, 'result', None)) for k, v in list(bk.name_map.items())} + print("NAMES:", names) + except Exception as e: + print("NAMES err", e) + for sh in bk.sheets(): + print("-" * 90) + print("SHEET", sh.name, "nrows", sh.nrows, "ncols", sh.ncols) + print("defcolwidth", sh.defcolwidth, "standardwidth", getattr(sh, 'standardwidth', None)) + print("colwidths:", [(c, sh.colinfo_map[c].width if c in sh.colinfo_map else None) for c in range(sh.ncols)]) + print("rowheights:", [(r + 1, sh.rowinfo_map[r].height if r in sh.rowinfo_map else None) for r in range(sh.nrows)]) + try: + pa = sh.phony_dim_rows + print("phony_dim_rows", pa) + except Exception: + pass + for attr in ("print_area", "print_titles"): + print(attr, getattr(bk, attr, None)) + +for p in sys.argv[1:]: + dump(p)