Compare commits

...
4 Commits
39 changed files with 1626 additions and 151 deletions
+3
View File
@@ -37,9 +37,12 @@ cd ..
安装开发验收依赖后,可执行 `python tools/browser_smoke.py --exe dist/博特维采购订单系统_单文件版.exe`。脚本在临时目录中复制并运行 EXE,验证页面、建单、手机尺寸审批、Excel 导入导出、带章 PDF 及重启后的归档一致性。EXE 测试不启用企业微信发送;测试数据和截图位于 `tools/out/ui-check-*`。
Windows 7 SP1 服务器使用单独的 [Win7 离线打包说明](packaging/win7/README_WIN7.md)。离线打包包包含 Python 3.8.10 官方安装程序、32/64 位离线依赖、两步构建脚本和自检;请在目标服务器生成 EXE。Win7 环境不要执行上面的 Python 3.11 打包命令。
## 主要能力
- 采购单和委外单,供应商/物料库、中文 XLS/XLSX 导入。
- 供应商支持维护公司简称,PDF 下载文件名采用“单据类型_订单号_供应商简称.pdf”;未填写简称时使用供应商全名,已审批文件重新下载也适用。
- 草稿可暂存,“保存并提交”保证当前修改一起进入审批。
- 数量、单价四位精度;逐行四舍五入到分再合计,页面、数据库和 PDF 使用同一口径。
- 制单与审批由不同账号完成,管理员也遵守分工;审批员一次确认即可通过或驳回。
Binary file not shown.
Binary file not shown.
+8
View File
@@ -0,0 +1,8 @@
@echo off
setlocal EnableExtensions DisableDelayedExpansion
set "PO_KIT_ROOT=%~dp0"
cd /d "%PO_KIT_ROOT%"
"%SystemRoot%\System32\WindowsPowerShell\v1.0\powershell.exe" -NoLogo -NoProfile -ExecutionPolicy Bypass -File "%PO_KIT_ROOT%win7\install_env.ps1" -ProbeOnly
set "PO_RESULT=%ERRORLEVEL%"
pause
exit /b %PO_RESULT%
+23
View File
@@ -0,0 +1,23 @@
@echo off
setlocal EnableExtensions DisableDelayedExpansion
set "PO_KIT_ROOT=%~dp0"
cd /d "%PO_KIT_ROOT%"
set "PYTHONUTF8=1"
set "PYTHONIOENCODING=utf-8"
set "PYTHONHOME="
set "PYTHONPATH="
"%SystemRoot%\System32\WindowsPowerShell\v1.0\powershell.exe" -NoLogo -NoProfile -ExecutionPolicy Bypass -File "%PO_KIT_ROOT%win7\install_env.ps1"
set "PO_RESULT=%ERRORLEVEL%"
if "%PO_RESULT%"=="3010" goto restart
if "%PO_RESULT%"=="1641" goto restart
if not "%PO_RESULT%"=="0" goto failed
echo Setup passed. Run 02_build_exe.cmd next.
goto finish
:restart
echo Windows requested a restart. Restart, then run 01_install_env.cmd again.
goto finish
:failed
echo SETUP FAILED. Exit code: %PO_RESULT%. Keep the logs folder and the message above.
:finish
pause
exit /b %PO_RESULT%
+21
View File
@@ -0,0 +1,21 @@
@echo off
setlocal EnableExtensions DisableDelayedExpansion
set "PO_KIT_ROOT=%~dp0"
cd /d "%PO_KIT_ROOT%"
set "PYTHONUTF8=1"
set "PYTHONIOENCODING=utf-8"
set "PYTHONHOME="
set "PYTHONPATH="
if not exist "%PO_KIT_ROOT%.buildenv\Scripts\python.exe" goto missing
"%PO_KIT_ROOT%.buildenv\Scripts\python.exe" "%PO_KIT_ROOT%win7\build.py" build
if errorlevel 1 goto failed
echo.
echo EXE is in the new dated subfolder under dist. See BUILD_INFO.json.
pause
exit /b 0
:missing
echo Run 01_install_env.cmd successfully first.
:failed
echo BUILD FAILED. Check the messages and build log. Keep the old server unchanged.
pause
exit /b 1
+78
View File
@@ -0,0 +1,78 @@
# Windows 7 服务器离线打包说明
本包用于在你的 Windows 7 服务电脑上,从 v4 源码生成 EXE。包括 Python 3.8.10 官方完整安装程序、32 位与 64 位离线依赖、构建脚本、环境与业务自检,以及现有 16 页使用升级手册。
安装脚本修复:新版解压到另一文件夹时,先检查本包及系统登记的 Python 3.8.10,确认位数和运行正常后复用;不再仅因新目录没有 `.python38` 就重复安装。只有找不到可用环境时才运行官方安装器。安装器错误码、依赖安装及自检输出会保存在 `logs/`。
2026-09-17 工作台更新:突出“需预付款”“已收票待付款”“已付款未交齐”“变更后待核对”。待跟进订单先按处理类别排列,同类较早订单优先;付款、交货、收票均完成后默认收起,仍可通过筛选查看。付款和交货分别显示三档状态;“未登记”和变更前待核对记录单独提示,不换算虚假的完成百分比。
在工作台或订单列表点击“登记收票 / 交货”“登记付款”,可直接在弹窗中保存,无需进入订单详情。制单员登记自己的采购跟进,审批员登记付款,管理员可登记两者。弹窗保存后保留当前筛选;如果其他人已更新订单,会提示重新读取并核对状态,避免覆盖。订单详情顶部显示最近经办人、时间和备注。手机端订单自动排成卡片。本次不增加业务字段,不改变已有审批或 PDF 归档;原 PDF 手册中工作台截图为旧布局,操作以本段说明及新版页面为准。
2026-09-17 文件名更新:在“供应商管理”的新增或编辑窗口填写“公司简称”,PDF 文件名即为“采购单_订单号_简称.pdf”(委外单使用对应类型名称);未填写简称时使用供应商全名。已审批订单重新下载也使用新文件名,归档 PDF 内容不变,无需重新审批。Excel 模板新增末列“公司简称”,旧表缺少该列时保留原简称。本次启动会先备份数据库,再升级到数据库结构版本 5,新增简称字段。
2026-09-17 排版更新:公司抬头从 12.5 磅加大到 20 磅,保持黑体居中。采购专用章直径增加 30%,固定在右侧需方签章空位;供方和需方签章区之间留出 20 毫米间隔。采购单与委外订单均已调整,示例见 `samples/`。本次 PDF 模板版本为 `po-v5`。
升级后新审批生成的正式 PDF 使用新排版。已审批归档的 PDF 保留原版;需要采用新版的订单,请通过系统的“变更重审”流程重新审批。使用本包重新解压、安装环境并生成 EXE 后,按下文备份和替换服务程序。
## 1. 先确认系统
- 在“计算机”上点右键 → “属性”,确认是 Windows 7 **Service Pack 1**,并查看系统类型是 32 位还是 64 位。脚本会自动按系统类型选择对应安装程序和依赖。
- Python 3.8 需要 Windows 7 的 KB2533623 或替代更新(如 KB3063858)所提供的 DLL 加载支持,以及 Universal C Runtime。安装 Python 时若提示系统补丁、运行库或重启,请先按提示补齐。补丁可能需要网管从微软获取,本包不自动更改系统补丁或防火墙。
- 建议把整个压缩包解压到较短的固定路径,例如 `C:\PO_Win7Build`。不要在压缩软件内直接运行脚本,也不要放到正在保存正式订单的目录中。
- 本包按 Win7 兼容依赖制作,当前电脑上的验证不等同于真实 Win7 验收;在目标服务器打包和自检通过后再投入使用。Win7 与 Python 3.8 已结束官方维护,按公司局域网范围部署。
## 2. 两步生成 EXE
1. 双击 `01_install_env.cmd`。
- 优先复用已安装且位数匹配的 Python 3.8.10;没有可用环境时才用包内安装器安装到 `.python38`。不改系统 PATH。
- 如果复用的是旧包文件夹内的 Python,请保留该文件夹,以便后续打包;生成的 EXE 运行不依赖该 Python 环境。
- 从本包 `wheels/x64` 或 `wheels/x86` 离线安装到 `.buildenv`,不连接 pip 源、不临时升级依赖。
- 自动检查密码哈希、中文 PDF、制单审批、部分/全部付款交货、Excel 台账和 HTTP 服务。自检仅使用临时数据库,通知不开启。
- 看到 `SETUP COMPLETE` 后关闭窗口。失败时查看并保留 `logs/setup-时间.log`;运行过安装器时还会有 `logs/python-installer-时间.log` 及其子日志。返回码 3010 或 1641 表示需要先重启电脑。
2. 双击 `02_build_exe.cmd`。
- 再次核对文件校验值、Python 与依赖版本,然后构建单文件程序。
- 打包期间可能数分钟没有新输出;详细日志在 `build/时间-位数/build.log`。
- 生成后自动在隔离目录启动一次 EXE,检查登录与初始化入口,并关闭该测试进程。
- 看到 `BUILD COMPLETE` 后,在 `dist/时间-位数/` 取出 **`PO_Server_Win7.exe`**。同目录还有 `BUILD_INFO.json` 和使用说明。
32 位 Windows 生成 32 位 EXE,64 位 Windows 生成 64 位 EXE。每次构建写到新的日期目录,不覆盖已有程序或业务数据。请保持本包原始源码及依赖不变,才能通过校验;需要修改源码时由维护人员重新制作本包。
## 3. 在服务器上使用
把生成的 EXE 放入固定服务目录,例如 `D:\PO_Server`,双击运行,按窗口显示的端口访问。首次部署时在此目录的 `instance/setup.token` 查看初始化口令。原系统升级时,先停机并完整备份,再替换程序,保留原 `instance`、`uploads`、`exports`;设置过 `PO_DATA_DIR` 的继续指向原数据目录。
同局域网的员工电脑和手机继续用浏览器访问服务电脑的内网 IP 与实际端口,例如 `http://192.168.1.100:8080`。建议在员工的较新电脑或手机上操作页面;服务电脑只负责运行 EXE,Win7 自带的 IE 不是本项目已验收浏览器。
首次正式使用前,至少用两个不同账号完成:创建测试订单 → 提交 → 审批盖章 → 下载中文 PDF → 登记部分和全部付款/交货 → 重启程序确认数据仍在。企业微信另按手册第 10-12 页配置并在真实公司 Wi-Fi 下验证。测试完成前,不删除旧程序和原数据备份。
## 4. 常见问题
| 现象 | 处理 |
| --- | --- |
| 旧包可以安装,新包刚开始就提示 SETUP FAILED | 使用这个修复包,保留原来可用的 Python 所在文件夹。先运行 `00_check_environment.cmd` 可查看识别到的 Python 路径;这个检查不安装依赖 |
| Python 已装好但自动没有找到 | 在包根目录新建 `python_path.txt`,仅填一行完整路径,例如 `D:\旧打包包\.python38\python.exe`,再运行第一个脚本。会检查是否为匹配位数的 Python 3.8.10,不需要复制旧的 `.buildenv` |
| Python 安装提示需要 SP1、更新或重启 | 先检查 Windows 7 SP1、DLL 加载更新及 UCRT。完成更新并重启,再运行第一个脚本 |
| 缺少 api-ms-win-crt、VCRUNTIME140、找不到程序入口点 | 检查系统运行库和更新;若自检提示具体 DLL,保留完整名称及错误信息。不要从陌生网站下载单个 DLL 覆盖系统目录 |
| 文件校验失败 | 重新完整解压本包;确认杀毒软件没有隔离文件、源码和 wheels 未被修改 |
| 路径过长、权限不足 | 重新解压到较短、可写的本地路径。不要移动已经建立的 `.buildenv`,在新目录重跑第一个脚本 |
| 自检密码或 PDF 失败 | 保留完整错误及依赖信息,暂不部署。包内检查包括新版账号 scrypt 哈希及本机中文字体加载 |
| 打包失败 | 查看新建的 `build/时间-位数/build.log`;不要改用普通 `requirements-build.txt`,它用于较新的 Windows/Python |
| EXE 自检失败 | 不替换正式服务;检查错误内容、系统补丁和杀毒隔离记录 |
| 本机可用,手机打不开 | 按手册核对实际 IP、端口、公司 Wi-Fi 和防火墙。通知配置中的地址也要同步 |
## 5. 兼容依赖与维护说明
本包使用主线 v4 业务源码,并包含上述 PDF 排版更新。兼容依赖完全固定,使用 Python 3.8.10、PyInstaller 5.13.2、Flask 3.0.3、Werkzeug 3.0.6、SQLAlchemy 2.0.44、ReportLab 4.2.5、Pillow 10.4.0 等。完整列表在 `win7/requirements-win7.txt`。
Waitress 使用本项目的 **3.0.2+win7.1** 兼容 wheel。HTTP 服务及安全修复代码与官方 3.0.2 保持逐字节一致,仅为命令行入口的名称解析增加 Python 3.8 回退,并修改版本元数据、重算 wheel 校验记录;原许可证保留。这样保留 3.0.1/3.0.2 的修复,没有降到旧的 3.0.0。上游官方支持范围仍为 Python 3.9+,这是本项目单独测试维护的兼容组合,不代表上游承诺 Win7 支持。原始 wheel 和重打包脚本随包提供。
`PACKAGE_INFO.json` 记录源代码提交与验证范围;`SHA256SUMS.json` 记录本包文件校验值。构建输出的 `BUILD_INFO.json` 记录实际操作系统、Python、位数和 EXE SHA-256。最终以你的服务器上的自检和现场流程验收为准。`00_check_environment.cmd` 只检查系统、可复用的 Python 并保存诊断日志,不安装软件或改系统设置。
## 官方参考
- Python 3.8.10 及官方安装程序:https://www.python.org/downloads/release/python-3810/
- Python 3.8 Windows 系统说明:https://docs.python.org/3.8/using/windows.html
- PyInstaller 5.13.2 的 Windows 构建与运行库说明:https://pyinstaller.org/en/v5.13.2/usage.html#windows
- Waitress 3.0.2 变更记录:https://docs.pylonsproject.org/projects/waitress/en/stable/
现有 PDF 手册描述 v4 常规 Windows 64 位版本的操作流程。**Win7 安装及打包环境以本文件为准**,无需按通用手册安装 Python 3.11 或 PyInstaller 6。
+122
View File
@@ -0,0 +1,122 @@
"""Offline setup and EXE build, intended to run on the Windows 7 server."""
import argparse
from datetime import datetime
import hashlib
import importlib.metadata
import json
import os
from pathlib import Path
import platform
import shutil
import struct
import subprocess
import sys
ROOT = Path(__file__).resolve().parents[1]
ARCH = 'x64' if struct.calcsize('P') == 8 else 'x86'
def run(args, **kwargs):
print('>', subprocess.list2cmdline([str(x) for x in args]), flush=True)
subprocess.run([str(x) for x in args], check=True, **kwargs)
def verify():
manifest = json.loads((ROOT/'SHA256SUMS.json').read_text(encoding='utf-8'))
for name, digest in manifest.items():
path = (ROOT/name).resolve()
if ROOT.resolve() not in path.parents:
raise RuntimeError('Invalid manifest path')
if hashlib.sha256(path.read_bytes()).hexdigest() != digest:
raise RuntimeError('File checksum mismatch: ' + name)
print('Package checksums verified.', flush=True)
def check_python():
if os.name != 'nt' or sys.version_info[:3] != (3, 8, 10):
raise RuntimeError('Use the bundled official Windows Python 3.8.10 installer first.')
version = sys.getwindowsversion()
if (version.major, version.minor) < (6, 1) or ((version.major, version.minor) == (6, 1) and version.service_pack_major < 1):
raise RuntimeError('Windows 7 SP1 or later is required.')
print('Python:', sys.version.split()[0], ARCH, '| OS:', platform.platform(), flush=True)
if (version.major, version.minor) != (6, 1):
print('NOTE: This is not Windows 7. The resulting EXE must still be built/tested on Windows 7.', flush=True)
def setup():
check_python()
verify()
env_dir = ROOT/'.buildenv'
python = env_dir/'Scripts/python.exe'
if not python.exists():
run([sys.executable, '-m', 'venv', env_dir])
wheels = ROOT/'wheels'/ARCH
options = ['--no-index', '--find-links', wheels, '--disable-pip-version-check', '--no-cache-dir']
run([python, '-m', 'pip', 'install'] + options + ['pip==25.0.1', 'setuptools==68.2.2', 'wheel==0.45.1'])
run([python, '-m', 'pip', 'install'] + options + ['-r', ROOT/'win7/requirements-win7.txt'])
run([python, '-m', 'pip', 'check'])
run([python, ROOT/'win7/selftest.py', ROOT/'po_app'])
print('SETUP COMPLETE. Run 02_build_exe.cmd next.', flush=True)
def build():
check_python()
verify()
# Running from the dedicated environment keeps modern packages out of the build.
expected = (ROOT/'.buildenv/Scripts/python.exe').resolve()
if Path(sys.executable).resolve() != expected:
raise RuntimeError('Run 02_build_exe.cmd after 01_install_env.cmd.')
for line in (ROOT/'win7/requirements-win7.txt').read_text().splitlines():
if not line or line.startswith('#'):
continue
package, pinned = line.split('==')
if importlib.metadata.version(package) != pinned:
raise RuntimeError('Dependency changed; run 01_install_env.cmd again: ' + package)
run([sys.executable, ROOT/'win7/selftest.py', ROOT/'po_app'])
build_id = datetime.now().strftime('%Y%m%d-%H%M%S-%f') + '-' + ARCH
work = ROOT/'build'/build_id
output = ROOT/'dist'/build_id
work.mkdir(parents=True)
output.mkdir(parents=True)
env = os.environ.copy()
env['PYINSTALLER_CONFIG_DIR'] = str(work/'cache')
command = [sys.executable, '-m', 'PyInstaller', '--noconfirm', '--onefile', '--console',
'--noupx', '--name', 'PO_Server_Win7', '--distpath', output,
'--workpath', work/'temp', '--specpath', work,
'--paths', ROOT/'po_app',
'--add-data', str(ROOT/'po_app/templates') + ';templates',
'--add-data', str(ROOT/'po_app/static') + ';static',
'--hidden-import', 'sqlalchemy.dialects.sqlite',
'--hidden-import', 'openpyxl', '--hidden-import', 'xlrd',
'--collect-all', 'reportlab', ROOT/'po_app/app.py']
log_path = work/'build.log'
print('Building EXE. Build log:', log_path, flush=True)
with log_path.open('wb') as log:
subprocess.run([str(x) for x in command], cwd=ROOT, env=env,
stdout=log, stderr=subprocess.STDOUT, check=True)
exe = output/'PO_Server_Win7.exe'
run([sys.executable, ROOT/'win7/probe_exe.py', exe])
info = {'built_on': platform.platform(), 'python': sys.version, 'architecture': ARCH,
'source_commit': json.loads((ROOT/'PACKAGE_INFO.json').read_text(encoding='utf-8'))['source_commit'],
'sha256': hashlib.sha256(exe.read_bytes()).hexdigest(),
'source_selftest': 'passed', 'exe_startup_probe': 'passed',
'built_on_windows7': sys.getwindowsversion()[:2] == (6, 1)}
(output/'BUILD_INFO.json').write_text(json.dumps(info, ensure_ascii=False, indent=2), encoding='utf-8')
shutil.copy2(ROOT/'README_WIN7.txt', output/'README_WIN7.txt')
for pdf in (ROOT/'manual').glob('*.pdf'):
shutil.copy2(pdf, output/pdf.name)
print('BUILD COMPLETE:', exe, flush=True)
print('Copy EXE to a fixed service folder. Keep the original instance/uploads/exports when upgrading.', flush=True)
if __name__ == '__main__':
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('command', choices=['setup', 'build', 'verify'])
args = parser.parse_args()
try:
{'setup': setup, 'build': build, 'verify': verify}[args.command]()
except Exception:
import traceback
traceback.print_exc()
print('FAILED. Do not replace the running server. Check the message and build log above.', flush=True)
sys.exit(1)
+98
View File
@@ -0,0 +1,98 @@
"""Assemble the offline Win7 source kit from verified downloads in a local cache."""
import argparse
from datetime import datetime
import hashlib
import json
from pathlib import Path
import shutil
import subprocess
import zipfile
REPO = Path(__file__).resolve().parents[2]
HERE = Path(__file__).resolve().parent
def assemble(cache, output):
cache, output = cache.resolve(), output.resolve()
if REPO not in output.parents:
raise ValueError('Output must be within this workspace')
if output.exists() or output.with_suffix('.zip').exists():
raise ValueError('Choose a new output directory; existing packages are not overwritten')
# Only committed application files are shipped, never runtime directories.
subprocess.run(['git', 'diff', '--exit-code', 'HEAD', '--', 'po_app'], cwd=REPO, check=True,
stdout=subprocess.DEVNULL)
commit = subprocess.check_output(['git', 'rev-parse', 'HEAD'], cwd=REPO, text=True).strip()
names = subprocess.check_output(['git', 'ls-files', '-z', 'po_app'], cwd=REPO).decode('utf-8').split('\0')
output.mkdir(parents=True)
def copy(source, relative):
dest = output/relative
dest.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(source, dest)
for name in names:
if not name:
continue
if name.endswith('.py') or name.startswith(('po_app/templates/', 'po_app/static/')):
copy(REPO/name, name)
for name in ['build.py', 'selftest.py', 'probe_exe.py', 'requirements-win7.txt', 'make_waitress_wheel.py', 'install_env.ps1']:
copy(HERE/name, 'win7/' + name)
for name in ['00_check_environment.cmd', '01_install_env.cmd', '02_build_exe.cmd']:
(output/name).write_bytes((HERE/name).read_text(encoding='utf-8').replace('\r\n', '\n').replace('\n', '\r\n').encode('ascii'))
(output/'README_WIN7.txt').write_text((HERE/'README_WIN7.md').read_text(encoding='utf-8'), encoding='utf-8-sig')
copy(REPO/'output/pdf/博特维采购订单系统_v4_使用与升级手册.pdf', 'manual/博特维采购订单系统_v4_使用与升级手册.pdf')
for sample in ('采购单_签章调整样张.pdf', '委外订单_签章调整样张.pdf'):
copy(REPO/'output/pdf'/sample, 'samples/' + sample)
for name, expected_md5 in [('python-3.8.10-amd64.exe', '62cf1a12a5276b0259e8761d4cf4fe42'),
('python-3.8.10.exe', 'b355cfc84b681ace8908ae50908e8761')]:
if hashlib.md5((cache/name).read_bytes()).hexdigest() != expected_md5:
raise ValueError('Installer differs from python.org download: ' + name)
copy(cache/name, 'installers/' + name)
pins = {}
for line in (HERE/'requirements-win7.txt').read_text().splitlines():
if line and not line.startswith('#'):
name, version = line.split('==')
pins[name.lower().replace('-', '_')] = version
for arch, folder in [('x64', 'wheels64'), ('x86', 'wheels32')]:
selected = []
for wheel in (cache/folder).glob('*.whl'):
name, version = wheel.name.split('-')[:2]
if pins.get(name.lower()) == version:
copy(wheel, 'wheels/' + arch + '/' + wheel.name)
selected.append(name.lower())
if set(selected) != set(pins) or len(selected) != len(pins):
raise ValueError('Incomplete wheel set: %s, missing %s' % (arch, set(pins)-set(selected)))
copy(cache/'waitress-3.0.2-py3-none-any.whl', 'third_party/waitress-3.0.2-py3-none-any.whl')
info = {'source_commit': commit, 'created_at': datetime.now().isoformat(timespec='seconds'),
'setup_revision': '20260917-reuse-installed-python',
'purpose': 'Build v4 on the target Windows 7 SP1 server',
'architectures': ['x86', 'x64'], 'python': '3.8.10',
'windows7_field_verified': False,
'validation_note': 'See VALIDATION.txt for local tests; target Win7 acceptance is still required.'}
(output/'PACKAGE_INFO.json').write_text(json.dumps(info, ensure_ascii=False, indent=2), encoding='utf-8')
if (cache/'VALIDATION.txt').exists():
copy(cache/'VALIDATION.txt', 'VALIDATION.txt')
hashes = {p.relative_to(output).as_posix(): hashlib.sha256(p.read_bytes()).hexdigest()
for p in sorted(output.rglob('*')) if p.is_file()}
(output/'SHA256SUMS.json').write_text(json.dumps(hashes, ensure_ascii=False, indent=2), encoding='utf-8')
archive = output.with_suffix('.zip')
with zipfile.ZipFile(archive, 'x', zipfile.ZIP_DEFLATED) as zipout:
for path in sorted(output.rglob('*')):
if path.is_file():
zipout.write(path, (Path(output.name)/path.relative_to(output)).as_posix())
with zipfile.ZipFile(archive) as zipout:
assert zipout.testzip() is None
assert all(zipout.read(output.name + '/' + name) == (output/name).read_bytes() for name in hashes)
digest = hashlib.sha256(archive.read_bytes()).hexdigest()
archive.with_suffix('.zip.sha256.txt').write_text(digest + ' ' + archive.name + '\n', encoding='ascii')
print(archive)
print('Bytes:', archive.stat().st_size, 'SHA256:', digest)
return archive
if __name__ == '__main__':
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--cache', type=Path, default=REPO/'tools/out/win7-prep')
parser.add_argument('--output', type=Path, required=True)
args = parser.parse_args()
assemble(args.cache, args.output)
+143
View File
@@ -0,0 +1,143 @@
# Compatible with Windows PowerShell 2.0. Reuse a verified full Python install.
param([switch]$ProbeOnly)
$ErrorActionPreference = 'Stop'
$kitRoot = Split-Path -Parent (Split-Path -Parent $MyInvocation.MyCommand.Path)
$logDirectory = Join-Path $kitRoot 'logs'
[void][IO.Directory]::CreateDirectory($logDirectory)
$runId = Get-Date -Format 'yyyyMMdd-HHmmss-fff'
$setupLog = Join-Path $logDirectory ('setup-' + $runId + '.log')
$utf8 = New-Object System.Text.UTF8Encoding($true)
[IO.File]::WriteAllText($setupLog, '', $utf8)
$env:PYTHONUTF8 = '1'
$env:PYTHONIOENCODING = 'utf-8'
$env:PYTHONHOME = $null
$env:PYTHONPATH = $null
[Console]::OutputEncoding = New-Object System.Text.UTF8Encoding($false)
$targetBits = 32
if ($env:PROCESSOR_ARCHITECTURE -eq 'AMD64' -or $env:PROCESSOR_ARCHITEW6432 -eq 'AMD64') {
$targetBits = 64
}
function Write-Log([string]$message) {
Write-Host $message
[IO.File]::AppendAllText($setupLog, $message + [Environment]::NewLine, $utf8)
}
function Invoke-Logged([string]$executable, [string[]]$arguments) {
# PowerShell 2 treats native stderr as ErrorRecord; preserve it in the log
# without mistaking harmless warnings for a failed process.
$savedPreference = $ErrorActionPreference
$ErrorActionPreference = 'Continue'
try {
$global:LASTEXITCODE = $null
& $executable @arguments 2>&1 | ForEach-Object { Write-Log ([string]$_) }
$code = $global:LASTEXITCODE
} finally {
$ErrorActionPreference = $savedPreference
}
if ($null -eq $code) { return 1 }
return $code
}
function Test-Python([string]$candidate) {
if (-not (Test-Path -LiteralPath $candidate -PathType Leaf)) { return $false }
Write-Log ('Checking Python: ' + $candidate)
$probe = 'import sys, struct, venv, ensurepip, ssl, sqlite3; print(sys.version); print(sys.executable); sys.exit(0 if sys.version_info[:3] == (3,8,10) and struct.calcsize(''P'')*8 == ' + $targetBits + ' else 3)'
$code = Invoke-Logged $candidate @('-I', '-X', 'utf8', '-c', $probe)
if ($code -ne 0) { Write-Log ('Python rejected; exit code: ' + $code) }
return ($code -eq 0)
}
function Find-Python {
$manualFile = Join-Path $kitRoot 'python_path.txt'
if (Test-Path -LiteralPath $manualFile) {
$manualPath = [IO.File]::ReadAllText($manualFile).Trim().Trim('"')
if (Test-Python $manualPath) { return $manualPath }
throw 'python_path.txt must contain the full path to a working Python 3.8.10 of the required architecture.'
}
$candidates = New-Object System.Collections.ArrayList
[void]$candidates.Add((Join-Path $kitRoot '.python38\python.exe'))
# Python's official installer registers the full interpreter, even when
# its files live in an older source kit directory.
foreach ($registryRoot in @('HKCU:\Software\Python\PythonCore',
'HKCU:\Software\WOW6432Node\Python\PythonCore',
'HKLM:\Software\Python\PythonCore',
'HKLM:\Software\WOW6432Node\Python\PythonCore')) {
foreach ($tag in @('3.8', '3.8-32')) {
$keyPath = Join-Path $registryRoot ($tag + '\InstallPath')
try {
if (Test-Path -LiteralPath $keyPath) {
$key = Get-Item -LiteralPath $keyPath
$executable = [string]$key.GetValue('ExecutablePath')
$directory = [string]$key.GetValue('')
if ($executable) { [void]$candidates.Add($executable) }
if ($directory) { [void]$candidates.Add((Join-Path $directory 'python.exe')) }
}
} catch {
Write-Log ('Registry lookup skipped: ' + $keyPath + ' : ' + $_.Exception.Message)
}
}
}
foreach ($candidate in ($candidates | Select-Object -Unique)) {
if (Test-Python $candidate) { return $candidate }
}
return $null
}
try {
Write-Log ('Setup log: ' + $setupLog)
Write-Log ('Package: ' + $kitRoot)
Write-Log ('Windows: ' + [Environment]::OSVersion.VersionString)
Write-Log ('PowerShell: ' + $PSVersionTable.PSVersion.ToString())
Write-Log ('Required Python: 3.8.10, ' + $targetBits + '-bit')
$python = Find-Python
if ($ProbeOnly) {
if ($python) { Write-Log ('FOUND reusable Python: ' + $python) }
else { Write-Log 'No working matching Python was found. No installation was attempted.' }
Write-Log 'DIAGNOSTIC COMPLETE. No dependencies or system settings were changed.'
exit 0
}
if (-not $python) {
$installerName = 'python-3.8.10.exe'
if ($targetBits -eq 64) { $installerName = 'python-3.8.10-amd64.exe' }
$installer = Join-Path $kitRoot ('installers\' + $installerName)
if (-not (Test-Path -LiteralPath $installer)) { throw ('Installer missing: ' + $installer) }
$installerLog = Join-Path $logDirectory ('python-installer-' + $runId + '.log')
$targetDirectory = Join-Path $kitRoot '.python38'
Write-Log ('No reusable Python found. Starting installer: ' + $installer)
Write-Log ('Installer log: ' + $installerLog)
$installerArgs = @('/passive', '/log', ('"' + $installerLog + '"'),
'InstallAllUsers=0', ('TargetDir="' + $targetDirectory + '"'),
'Include_launcher=0', 'InstallLauncherAllUsers=0', 'PrependPath=0',
'Include_doc=0', 'Include_test=0', 'Include_tcltk=0', 'Shortcuts=0', 'AssociateFiles=0')
$process = Start-Process -FilePath $installer -ArgumentList $installerArgs -Wait -PassThru -WindowStyle Hidden
$installerCode = $process.ExitCode
Write-Log ('Python installer exit code: ' + $installerCode)
if ($installerCode -eq 3010 -or $installerCode -eq 1641) {
Write-Log 'Windows requested a restart. Restart the computer and run 01_install_env.cmd again.'
exit $installerCode
}
if ($installerCode -ne 0) {
throw ('Python installer failed with exit code ' + $installerCode + '. See ' + $installerLog)
}
# Maintenance mode may leave Python in its original registered folder.
$python = Find-Python
if (-not $python) {
throw 'Installer returned success but no matching Python was found. Keep logs; do not uninstall the old working environment.'
}
}
Write-Log ('Using existing Python: ' + $python)
Write-Log 'Keep this Python directory while using the build environment.'
$buildScript = Join-Path $kitRoot 'win7\build.py'
$code = Invoke-Logged $python @('-I', '-X', 'utf8', $buildScript, 'setup')
Write-Log ('Environment setup exit code: ' + $code)
if ($code -ne 0) { exit $code }
Write-Log 'SETUP COMPLETE. Run 02_build_exe.cmd next.'
exit 0
} catch {
Write-Log ('FAILED: ' + $_.Exception.Message)
Write-Log ('Details: ' + $_.ScriptStackTrace)
Write-Log ('Keep the logs folder. Setup log: ' + $setupLog)
exit 1
}
+60
View File
@@ -0,0 +1,60 @@
"""Repackage official Waitress 3.0.2 metadata for the tested Python 3.8 build.
The HTTP server and security fixes are kept byte-for-byte. Only the CLI name
resolver gets a Python 3.8 fallback. The local version identifies the adaptation.
"""
import base64
import csv
import hashlib
import io
from pathlib import Path
import zipfile
UPSTREAM_SHA256 = 'c56d67fd6e87c2ee598b76abdd4e96cfad1f24cacdea5078d382b1f9d7b5ed2e'
VERSION = '3.0.2+win7.1'
def repackage(source, destination):
source, destination = Path(source), Path(destination)
if hashlib.sha256(source.read_bytes()).hexdigest() != UPSTREAM_SHA256:
raise ValueError('Unexpected upstream Waitress wheel checksum')
old_info, new_info = 'waitress-3.0.2.dist-info', 'waitress-%s.dist-info' % VERSION
destination.mkdir(parents=True, exist_ok=True)
target = destination / ('waitress-%s-py3-none-any.whl' % VERSION)
records = []
with zipfile.ZipFile(source) as src, zipfile.ZipFile(target, 'w', zipfile.ZIP_DEFLATED) as out:
for name in src.namelist():
if name.endswith('/RECORD'):
continue
content = src.read(name)
new_name = name.replace(old_info, new_info)
if name == 'waitress/runner.py':
assert b'import pkgutil\n' in content and b'pkgutil.resolve_name(args[0])' in content
content = content.replace(b'import pkgutil\n',
b'try:\n from pkgutil import resolve_name\nexcept ImportError:\n from pkgutil_resolve_name import resolve_name\n')
content = content.replace(b'pkgutil.resolve_name(args[0])', b'resolve_name(args[0])')
if name == old_info + '/METADATA':
content = content.replace(b'Version: 3.0.2\n', ('Version: %s\n' % VERSION).encode())
content = content.replace(b'Requires-Python: >=3.9.0\n',
b'Requires-Python: >=3.8.0\nRequires-Dist: pkgutil-resolve-name==1.3.10; python_version < "3.9"\n')
assert b'Requires-Python: >=3.8.0\n' in content
out.writestr(new_name, content)
digest = base64.urlsafe_b64encode(hashlib.sha256(content).digest()).rstrip(b'=').decode()
records.append((new_name, 'sha256=' + digest, str(len(content))))
records.append((new_info + '/RECORD', '', ''))
record = io.StringIO(newline='')
csv.writer(record, lineterminator='\n').writerows(records)
out.writestr(new_info + '/RECORD', record.getvalue())
with zipfile.ZipFile(source) as src, zipfile.ZipFile(target) as dest:
assert all(src.read(name) == dest.read(name) for name in src.namelist()
if name.startswith('waitress/') and name != 'waitress/runner.py')
return target
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('source', type=Path)
parser.add_argument('destination', type=Path)
args = parser.parse_args()
print(repackage(args.source, args.destination))
+49
View File
@@ -0,0 +1,49 @@
"""Start the built EXE in isolation, verify setup/login, then stop its own tree."""
import os
from pathlib import Path
import socket
import subprocess
import tempfile
import time
import urllib.request
def probe(executable):
with tempfile.TemporaryDirectory(prefix='po-win7-exe-') as temporary:
directory = Path(temporary)
with socket.socket() as sock:
sock.bind(('127.0.0.1', 0))
port = sock.getsockname()[1]
env = {k: v for k, v in os.environ.items()
if not k.upper().startswith(('PO_', 'PYTHON', 'VIRTUAL_ENV'))}
env.update(PO_DATA_DIR=str(directory), PO_HOST='127.0.0.1', PO_OPEN_BROWSER='0')
http = urllib.request.build_opener(urllib.request.ProxyHandler({}))
with (directory/'server.log').open('wb') as log:
process = subprocess.Popen([str(Path(executable).resolve()), str(port)], env=env,
cwd=directory, stdout=log, stderr=subprocess.STDOUT,
creationflags=subprocess.CREATE_NO_WINDOW)
try:
deadline = time.monotonic() + 90
while time.monotonic() < deadline:
if process.poll() is not None:
raise RuntimeError('EXE exited: ' + (directory/'server.log').read_text(errors='replace'))
try:
with http.open('http://127.0.0.1:%s/login' % port, timeout=1) as response:
if response.status == 200:
assert (directory/'instance/setup.token').is_file()
print('EXE STARTUP PASSED:', executable)
return
except (OSError, TimeoutError):
time.sleep(0.25)
raise RuntimeError('EXE did not start within 90 seconds')
finally:
if process.poll() is None:
subprocess.run(['taskkill', '/PID', str(process.pid), '/T', '/F'],
check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
creationflags=subprocess.CREATE_NO_WINDOW)
process.wait(timeout=20)
if __name__ == '__main__':
import sys
probe(sys.argv[1])
+32
View File
@@ -0,0 +1,32 @@
# Windows 7 compatibility environment: CPython 3.8.10 only.
# The local Waitress wheel adapts only the CLI resolver; see README.
Flask==3.0.3
Werkzeug==3.0.6
Jinja2==3.1.6
MarkupSafe==2.1.5
click==8.1.8
itsdangerous==2.2.0
blinker==1.8.2
colorama==0.4.6
SQLAlchemy==2.0.44
greenlet==3.1.1
typing_extensions==4.13.2
reportlab==4.2.5
pillow==10.4.0
chardet==5.2.0
openpyxl==3.1.5
et-xmlfile==2.0.0
xlrd==2.0.2
waitress==3.0.2+win7.1
pkgutil-resolve-name==1.3.10
importlib-metadata==8.5.0
zipp==3.20.2
PyInstaller==5.13.2
pyinstaller-hooks-contrib==2023.8
pefile==2023.2.7
pywin32-ctypes==0.2.3
altgraph==0.17.4
setuptools==68.2.2
wheel==0.45.1
pip==25.0.1
packaging==24.2
+120
View File
@@ -0,0 +1,120 @@
"""Exercise the real application with isolated data; never send notifications."""
import argparse
from datetime import date
import io
import os
from pathlib import Path
import re
import secrets
import sys
import tempfile
import threading
import urllib.request
def run(source, artifacts=None):
import openpyxl
from PIL import Image
from waitress import create_server
from werkzeug.security import generate_password_hash, check_password_hash
password = 'Win7-Test-' + secrets.token_hex(10)
hashed = generate_password_hash(password)
assert check_password_hash(hashed, password), 'Password hashing failed'
# Also test the scrypt parameters generated by newer Windows deployments.
assert check_password_hash(generate_password_hash(password, method='scrypt:32768:8:1'), password)
Image.new('RGB', (3, 3)).save(io.BytesIO(), format='PNG')
with tempfile.TemporaryDirectory(prefix='po-win7-check-') as temporary:
os.environ['PO_DATA_DIR'] = temporary
os.environ.pop('PO_SECRET_KEY', None)
os.environ.pop('PO_HTTPS', None)
sys.path.insert(0, str(source.resolve()))
import app as module
from models import User
module.app.config['TESTING'] = True
assert (Path(temporary)/'instance/setup.token').is_file()
with module.Session() as session:
for name, role in [('maker', 'maker'), ('admin', 'admin')]:
session.add(User(username=name, name=name, role=role, active=True,
must_change_password=False, session_version=1, password_hash=hashed))
session.commit()
def post(client, path, data):
client.get('/login')
with client.session_transaction() as session:
token = session['csrf_token']
return client.post(path, data=dict(data, csrf_token=token))
clients = {}
for name in ('maker', 'admin'):
client = module.app.test_client()
assert post(client, '/login', {'username': name, 'password': password}).status_code == 302
clients[name] = client
def version(oid):
with module.Session() as session:
return session.get(module.Order, oid).row_version
def action(actor, oid, endpoint, **fields):
fields['row_version'] = str(version(oid))
return post(clients[actor], '/orders/%d/%s' % (oid, endpoint), fields)
try:
for kind in ('purchase', 'outsourcing'):
response = post(clients['maker'], '/orders/new',
{'doc_type': kind, 'request_key': secrets.token_hex(16)})
assert response.status_code == 302
oid = int(re.search(r'/orders/(\d+)', response.location).group(1))
response = action('maker', oid, 'edit', supplier_name='Win7 test supplier',
our_company='苏州博特维智能科技有限公司', payment_terms='月结',
delivery_place='苏州', order_date=date.today().isoformat(),
item_code='TEST-001', item_spec='测试物料', item_unit='件',
item_qty='2', item_price='12.3456', item_date=date.today().isoformat(),
item_remark='', intent='submit')
assert response.status_code == 302, response.get_data(as_text=True)
assert action('maker', oid, 'approve', stamped='1').status_code == 403
assert action('admin', oid, 'approve', stamped='1').status_code == 302
pdf = clients['maker'].get('/orders/%d/pdf' % oid).data
assert pdf.startswith(b'%PDF-') and b'%%EOF' in pdf
if artifacts:
artifacts.mkdir(parents=True, exist_ok=True)
(artifacts/(kind + '.pdf')).write_bytes(pdf)
for payment, delivery in [('partial', 'partial'), ('paid', 'delivered')]:
assert action('maker', oid, 'follow-up', needs_prepayment='1',
invoice_status='received', delivery_status=delivery,
fulfillment_note='离线环境自检', reason='').status_code == 302
assert action('admin', oid, 'payment', payment_status=payment,
payment_date=date.today().isoformat(), payment_note='', reason='').status_code == 302
assert clients['maker'].get('/orders/%d/pdf' % oid).data == pdf
print('PASS:', kind, 'login, submit, approval, PDF, partial/full follow-up, archive')
workbook = openpyxl.load_workbook(io.BytesIO(clients['admin'].get('/export/ledger').data))
assert workbook.active.max_row == 3
workbook.close()
assert clients['admin'].get('/orders?tracking=completed').status_code == 200
assert clients['admin'].get('/settings/notifications').status_code == 200
# Test a real HTTP socket using the exact Waitress compatibility wheel.
server = create_server(module.app, host='127.0.0.1', port=0, threads=2)
thread = threading.Thread(target=server.run, daemon=True)
thread.start()
try:
http = urllib.request.build_opener(urllib.request.ProxyHandler({}))
with http.open('http://127.0.0.1:%s/login' % server.effective_port, timeout=10) as response:
assert response.status == 200
response.read()
finally:
server.task_dispatcher.shutdown()
server.close()
thread.join(timeout=10)
print('PASS: Excel, dashboard, notification settings, Waitress HTTP')
finally:
module.engine.dispose()
print('SELFTEST PASSED. Only temporary data was used. No group messages sent.')
if __name__ == '__main__':
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('source', type=Path)
parser.add_argument('--artifacts', type=Path)
args = parser.parse_args()
run(args.source, args.artifacts)
+29 -10
View File
@@ -8,13 +8,14 @@ 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 sqlalchemy import func, or_
from werkzeug.security import check_password_hash, generate_password_hash
from security import db, current_user, login_required, role_required, init_security, validate_password, sign_in
from money import number, format_price
from order_logic import scope
from order_routes import register_order_routes
from order_tracking import TRACKING_FILTERS, filter_tracking, register_tracking_routes
from order_tracking import (TRACKING_FILTERS, filter_tracking, register_tracking_routes,
tracking_conditions, tracking_priority)
from excel_io import read_rows, read_material_rows, read_supplier_rows, save_upload
from flask import abort
@@ -77,10 +78,14 @@ def index():
month_total = (scope(db.query(func.sum(Order.total_amount)))
.filter(Order.status == "approved").scalar() or 0)
tracking_counts = {key: filter_tracking(scope(db.query(Order)), key).count() for key in TRACKING_FILTERS}
tracking_rows = scope(db.query(Order)).filter_by(status='approved').order_by(Order.id.desc()).limit(12).all()
tracking_key = request.args.get('tracking', 'open')
tracking_query = filter_tracking(scope(db.query(Order)), tracking_key or 'open')
tracking_total = tracking_query.count()
tracking_rows = tracking_query.order_by(tracking_priority(), Order.id.asc()).limit(12).all()
return render_template("index.html", user=u, todo=todo, my_orders=my_orders,
counts=counts, month_total=month_total,
tracking_counts=tracking_counts, tracking_rows=tracking_rows,
tracking_key=tracking_key or 'open', tracking_total=tracking_total,
status_text=STATUS_TEXT, role_text=ROLE_TEXT)
@@ -108,7 +113,10 @@ def orders():
q = q.filter(Order.order_date <= date_to)
if u.role == "maker":
q = q.filter(Order.created_by == u.id)
q = filter_tracking(q, request.args.get('tracking', ''))
tracking = request.args.get('tracking', '')
q = filter_tracking(q, tracking)
if not tracking and request.args.get('show_completed') != '1':
q = q.filter(or_(Order.status != 'approved', ~tracking_conditions()['closed']))
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)
@@ -147,6 +155,11 @@ def suppliers():
setattr(s, f, request.form.get(f, "").strip())
if not s.name:
raise ValueError('供应商名称必填')
if 'short_name' in request.form:
short_name = request.form['short_name'].strip()
if len(short_name) > 60:
raise ValueError('供应商简称不能超过60个字符')
s.short_name = short_name
if not sid:
db.add(s)
db.commit()
@@ -208,12 +221,16 @@ def supplier_import_excel():
if not any(r):
continue
(code, name, contact, phone, payment_terms, tax_no,
bank_name, bank_account) = (list(r) + [""] * 8)[:8]
bank_name, bank_account, short_name) = (list(r) + [None] * 9)[:9]
code, name = s(code), s(name)
if not name:
failed += 1
errs.append(u"第%d行:供应商名称为空" % i)
continue
if short_name is not None and len(s(short_name)) > 60:
failed += 1
errs.append(u"第%d行:供应商简称不能超过60个字符" % 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()
@@ -229,13 +246,15 @@ def supplier_import_excel():
obj.tax_no = s(tax_no)
obj.bank_name = s(bank_name)
obj.bank_account = s(bank_account)
if short_name is not None:
obj.short_name = s(short_name)
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))
bank_account=s(bank_account), short_name=s(short_name), active=True))
added += 1
db.commit()
msg = u"导入完成:新增 %d,更新 %d,跳过 %d,失败 %d" % (added, updated, skipped, failed)
@@ -254,11 +273,11 @@ def supplier_import_template():
wb = openpyxl.Workbook()
ws = wb.active
ws.append([u"编号", u"名称", u"联系人", 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)):
"4562 1088 0012 3456", u"精锐五金"])
for col, w in zip("ABCDEFGHI", (10, 30, 12, 16, 14, 22, 26, 24, 18)):
ws.column_dimensions[col].width = w
buf = io.BytesIO()
wb.save(buf)
@@ -406,7 +425,7 @@ def api_materials():
@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,
return jsonify([{"id": s.id, "name": s.name, "short_name": s.short_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])
+6 -5
View File
@@ -66,7 +66,7 @@ def read_rows(path):
return out
def read_table(path, keys, order):
def read_table(path, keys, order, missing=""):
"""通用表格读取:按关键词映射表头,返回按 order 排列的二维列表。
与 read_rows() 的区别是表头判定更宽松(命中任意关键词即可),
@@ -110,7 +110,7 @@ def read_table(path, keys, order):
vals = []
for k in order:
i = mapping.get(k)
vals.append(row[i] if (i is not None and i < len(row)) else "")
vals.append(row[i] if (i is not None and i < len(row)) else missing)
out.append(vals)
return out
@@ -124,6 +124,7 @@ MATERIAL_KEYS = {
MATERIAL_ORDER = ["code", "spec", "unit", "price"]
SUPPLIER_KEYS = {
"short_name": [u"简称", u"别名"],
"code": [u"编号", u"供应商编号", u"代码"],
"name": [u"名称", u"供应商名称", u"公司"],
"contact": [u"联系人"],
@@ -134,7 +135,7 @@ SUPPLIER_KEYS = {
"bank_account": [u"银行帐号", u"银行账号", u"帐号", u"账号"],
}
SUPPLIER_ORDER = ["code", "name", "contact", "phone", "payment_terms",
"tax_no", "bank_name", "bank_account"]
"tax_no", "bank_name", "bank_account", "short_name"]
def read_material_rows(path):
@@ -143,5 +144,5 @@ def read_material_rows(path):
def read_supplier_rows(path):
"""读取供应商表,返回 [编号,名称,联系人,电话,付款方式,税号,开户银行,银行帐号]。"""
return read_table(path, SUPPLIER_KEYS, SUPPLIER_ORDER)
"""简称追加在末列;旧表缺少该列时以 None 表示,更新时保留已有简称。"""
return read_table(path, SUPPLIER_KEYS, SUPPLIER_ORDER, missing=None)
+8 -3
View File
@@ -14,7 +14,7 @@ def upgrade_database(path):
if 'orders' not in tables:
return
version = con.execute('PRAGMA user_version').fetchone()[0]
if version >= 4:
if version >= 5:
return
duplicates = con.execute('SELECT order_no, COUNT(*) FROM orders GROUP BY order_no HAVING COUNT(*) > 1').fetchall()
if duplicates:
@@ -28,8 +28,13 @@ def upgrade_database(path):
upgrade_fixed_amounts(con)
if version < 3:
merge_approval(con, tables)
add_tracking(con)
con.execute('PRAGMA user_version=4')
if version < 4:
add_tracking(con)
if 'suppliers' in tables:
columns = {row[1] for row in con.execute('PRAGMA table_info(suppliers)')}
if 'short_name' not in columns:
con.execute("ALTER TABLE suppliers ADD COLUMN short_name VARCHAR(60) NOT NULL DEFAULT ''")
con.execute('PRAGMA user_version=5')
con.commit()
except Exception:
con.rollback()
+2 -1
View File
@@ -81,6 +81,7 @@ class Supplier(Base):
id = Column(Integer, primary_key=True)
code = Column(String(50))
name = Column(String(200), nullable=False)
short_name = Column(String(60), nullable=False, default='')
contact = Column(String(50))
phone = Column(String(50))
tax_no = Column(String(80))
@@ -277,5 +278,5 @@ def init_db(db_path):
Base.metadata.create_all(engine)
if not existed:
with engine.connect() as con:
con.exec_driver_sql('PRAGMA user_version=4')
con.exec_driver_sql('PRAGMA user_version=5')
return engine, sessionmaker(bind=engine)
+32 -3
View File
@@ -1,6 +1,8 @@
import hashlib
import io
import json
import re
import unicodedata
from datetime import datetime
from flask import abort, flash, redirect, render_template, request, send_file, url_for
@@ -13,7 +15,7 @@ from order_logic import (ORDER_FIELDS, ITEM_FIELDS, add_log, can_edit, can_appro
can_reject, clear_approval, creation_request, get_order, make_item,
next_order_no, recalc, remember_creation, save_form, snapshot, submit,
validate_order)
from pdf_export import build_pdf, seal_layout
from pdf_export import PDF_TEMPLATE_VERSION, build_pdf, seal_layout
from order_tracking import can_follow, can_record_payment
from security import current_user, db, login_required, role_required
@@ -24,6 +26,32 @@ def pdf_bytes(order):
return buf.getvalue()
def pdf_download_name(order):
# Use current master-data aliases without regenerating an approved PDF.
company = (order.supplier_name or '').strip()
matches = db.query(Supplier.short_name, Supplier.tax_no).filter(Supplier.name == company).all()
tax_no = (order.supplier_tax_no or '').strip()
if tax_no:
same_tax = [row for row in matches if (row.tax_no or '').strip() == tax_no]
if same_tax:
matches = same_tax
aliases = {(row.short_name or '').strip() or company for row in matches}
# Ambiguous duplicate names fall back to the company recorded on the order.
supplier = next(iter(aliases)) if len(aliases) == 1 else company
def part(value, fallback, limit):
value = ''.join(c for c in str(value or '') if not unicodedata.category(c).startswith('C'))
value = re.sub(r'[<>:"/\\|?*]', '_', value)
value = re.sub(r'\s+', ' ', value).strip(' .')
# Windows counts UTF-16 units; retain Chinese and bound the full filename.
value = value.encode('utf-16-le')[:limit * 2].decode('utf-16-le', errors='ignore').rstrip(' .')
return value or fallback
return '%s_%s_%s.pdf' % (part(order.doc_type_text, '订单', 32),
part(order.order_no, '未编号', 64),
part(supplier, '未填写供应商', 80))
def register_order_routes(app):
@app.route('/orders/new', methods=['GET', 'POST'])
@login_required
@@ -131,7 +159,8 @@ def register_order_routes(app):
version = db.query(OrderVersion).filter_by(order_id=oid, revision=order.revision).one()
version.pdf_data = pdf_bytes(order)
version.pdf_sha256 = hashlib.sha256(version.pdf_data).hexdigest()
version.template_version = 'po-v3-legacy-current' if version.template_version == 'legacy-current' else 'po-v3'
version.template_version = (PDF_TEMPLATE_VERSION + '-legacy-current'
if version.template_version == 'legacy-current' else PDF_TEMPLATE_VERSION)
add_log(order, '审批通过并归档正式 PDF', user, request.form.get('comment'), old)
db.commit()
flash('审批通过,正式 PDF 已归档', 'ok')
@@ -253,7 +282,7 @@ def register_order_routes(app):
else:
data = pdf_bytes(order)
response = send_file(io.BytesIO(data), mimetype='application/pdf', as_attachment=request.args.get('dl') == '1',
download_name='%s %s.pdf' % (order.doc_type_text, order.order_no))
download_name=pdf_download_name(order))
response.headers['Cache-Control'] = 'private, no-store'
return response
+69 -7
View File
@@ -1,8 +1,8 @@
"""Internal order follow-up; approved documents stay frozen."""
from datetime import date
from flask import abort, flash, redirect, request, url_for
from sqlalchemy import and_, or_
from flask import abort, flash, jsonify, redirect, render_template, request, url_for
from sqlalchemy import and_, case, or_
from models import DELIVERY_STATUS, INVOICE_STATUS, PAYMENT_STATUS, Order
from order_logic import add_log, get_order, normalize_date
@@ -10,6 +10,7 @@ from security import current_user, db, login_required
TRACKING_FILTERS = {
'open': '待跟进订单',
'prepayment': '需预付款',
'invoice_due': '已收票待付款',
'unpaid': '未付款',
@@ -22,6 +23,7 @@ TRACKING_FILTERS = {
'completed': '已付款且已交货',
'unrecorded': '跟进待登记',
'recheck': '变更后待核对',
'closed': '付款、交货、收票均完成',
}
@@ -30,7 +32,12 @@ def tracking_conditions():
current_payment = or_(Order.payment_revision < 0, Order.payment_revision == Order.revision)
payment_due = and_(current_payment, Order.payment_status.in_(('unpaid', 'partial')))
delivery_open = and_(current_fulfillment, Order.delivery_status.in_(('pending', 'partial')))
completed = and_(current_payment, current_fulfillment, Order.payment_status == 'paid',
Order.delivery_status == 'delivered')
closed = and_(completed, Order.invoice_status == 'received')
return {
'open': ~closed,
'closed': closed,
'prepayment': and_(current_fulfillment, Order.needs_prepayment.is_(True), payment_due),
'invoice_due': and_(current_fulfillment, Order.invoice_status == 'received', payment_due),
'unpaid': and_(current_payment, Order.payment_status == 'unpaid'),
@@ -40,8 +47,7 @@ def tracking_conditions():
'partial_delivery': and_(current_fulfillment, Order.delivery_status == 'partial'),
'delivered': and_(current_fulfillment, Order.delivery_status == 'delivered'),
'paid_undelivered': and_(current_payment, Order.payment_status == 'paid', delivery_open),
'completed': and_(current_payment, current_fulfillment, Order.payment_status == 'paid',
Order.delivery_status == 'delivered'),
'completed': completed,
'unrecorded': or_(Order.invoice_status == 'unknown', Order.delivery_status == 'unknown',
Order.payment_status == 'unknown'),
'recheck': or_(~current_fulfillment, ~current_payment),
@@ -57,6 +63,48 @@ def filter_tracking(query, key):
return query.filter(Order.status == 'approved', conditions[key])
def tracking_priority():
conditions = tracking_conditions()
return case(*[(conditions[key], rank) for rank, key in enumerate(
('recheck', 'prepayment', 'invoice_due', 'paid_undelivered', 'unrecorded'))], else_=5)
def tracking_tasks(order):
if order.status != 'approved':
return []
if order.fulfillment_stale or order.payment_stale:
return [('recheck', '变更后待核对')]
tasks = []
if order.payment_status in ('unpaid', 'partial'):
if order.needs_prepayment:
tasks.append(('prepayment', '安排预付款' if order.payment_status == 'unpaid' else '核对预付款及剩余付款安排'))
elif order.invoice_status == 'received':
tasks.append(('invoice_due', '已收票,跟进付款' if order.payment_status == 'unpaid' else '跟进剩余付款'))
else:
tasks.append(('payment', '跟进付款安排'))
if order.delivery_status in ('pending', 'partial'):
tasks.append(('delivery', '跟进剩余交货' if order.delivery_status == 'partial' else '跟进交货'))
if order.invoice_status == 'pending':
tasks.append(('invoice', '跟进收票'))
if 'unknown' in (order.payment_status, order.delivery_status, order.invoice_status):
tasks.append(('unrecorded', '补充跟进登记'))
return tasks or [('closed', '付款、交货、收票均完成')]
def latest_tracking(order):
latest = {}
for log in reversed(order.logs):
if log.action in ('更新采购跟进', '登记付款状态') and log.action not in latest:
latest[log.action] = log
return latest
def tracking_response(oid):
if request.headers.get('X-Tracking-Modal') == '1':
return jsonify(ok=True)
return redirect(url_for('order_view', oid=oid) + '#order-tracking')
def can_follow(order, user):
return (order.status == 'approved' and user.role in ('maker', 'admin')
and (user.role == 'admin' or order.created_by == user.id))
@@ -84,7 +132,21 @@ def register_tracking_routes(app):
@app.context_processor
def tracking_context():
return {'INVOICE_STATUS': INVOICE_STATUS, 'DELIVERY_STATUS': DELIVERY_STATUS,
'PAYMENT_STATUS': PAYMENT_STATUS, 'TRACKING_FILTERS': TRACKING_FILTERS}
'PAYMENT_STATUS': PAYMENT_STATUS, 'TRACKING_FILTERS': TRACKING_FILTERS,
'can_follow': can_follow, 'can_record_payment': can_record_payment,
'tracking_tasks': tracking_tasks, 'latest_tracking': latest_tracking}
@app.get('/orders/<int:oid>/tracking/<kind>')
@login_required
def tracking_editor(oid, kind):
if kind not in ('follow-up', 'payment'):
abort(404)
order, user = get_order(oid), current_user()
allowed = can_follow(order, user) if kind == 'follow-up' else can_record_payment(order, user)
if not allowed:
abort(403)
template = 'tracking_editor_fragment.html' if request.args.get('fragment') == '1' else 'tracking_editor.html'
return render_template(template, o=order, user=user, kind=kind)
@app.post('/orders/<int:oid>/follow-up')
@login_required
@@ -124,7 +186,7 @@ def register_tracking_routes(app):
add_log(order, '更新采购跟进', user, ';'.join(changes + (['更正原因:' + reason] if reason else [])))
db.commit()
flash('采购跟进已保存', 'ok')
return redirect(url_for('order_view', oid=oid) + '#order-tracking')
return tracking_response(oid)
@app.post('/orders/<int:oid>/payment')
@login_required
@@ -160,4 +222,4 @@ def register_tracking_routes(app):
add_log(order, '登记付款状态', user, ';'.join(changes + (['更正原因:' + reason] if reason else [])))
db.commit()
flash('付款状态已保存', 'ok')
return redirect(url_for('order_view', oid=oid) + '#order-tracking')
return tracking_response(oid)
+52 -23
View File
@@ -60,6 +60,8 @@ STATUS_WATERMARK = {
SEAL_COLOR = colors.HexColor("#C8102E")
SEAL_NUMBER = "3205070994140"
PDF_TEMPLATE_VERSION = "po-v5"
PDF_SEAL_SIZE = 32 * 1.3 * mm
def _seal_arc(text, radius, start, span, font, size, bottom=False,
@@ -112,7 +114,7 @@ def seal_layout(company, seal_number=SEAL_NUMBER):
class SealFlowable(Flowable):
"""公司名和印章号码环绕一圈的采购章,不含个人及日期。"""
def __init__(self, company, title, seal_number=SEAL_NUMBER, size=32 * mm):
def __init__(self, company, title, seal_number=SEAL_NUMBER, size=PDF_SEAL_SIZE):
Flowable.__init__(self)
self.layout = seal_layout(company, seal_number)
self.title = title or u"采购专用章"
@@ -153,6 +155,38 @@ class SealFlowable(Flowable):
c.restoreState()
class ConfirmationFlowable(Flowable):
"""Keep each party's label and signing space in the same fixed-size cell."""
def __init__(self, party, width, seal=None):
Flowable.__init__(self)
self.label = u"%s确认(盖章):" % party
self.width = width
self.height = PDF_SEAL_SIZE + 4 * mm
self.seal = seal
def draw(self):
c = self.canv
c.saveState()
font_size = 9.5
center_y = self.height / 2
baseline = center_y - font_size * 0.35
c.setFont(FONT_BODY, font_size)
c.setFillColor(colors.black)
c.drawString(0, baseline, self.label)
signing_start = pdfmetrics.stringWidth(self.label, FONT_BODY, font_size) + 2 * mm
signing_end = self.width - 1 * mm
if self.seal:
# Center the seal in the buyer's signing space, next to its label.
self.seal.drawOn(c, (signing_start + signing_end - self.seal.width) / 2,
center_y - self.seal.height / 2)
else:
c.setStrokeColor(colors.HexColor("#888888"))
c.setLineWidth(0.4)
c.line(signing_start, baseline - 1.5, signing_end, baseline - 1.5)
c.restoreState()
def _find_font(candidates):
for d in FONT_DIRS:
for name in candidates:
@@ -244,7 +278,7 @@ def build_pdf(order, items, out_path, watermark=None):
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,
"company": ParagraphStyle("company", fontName=FONT_BOLD, fontSize=20, leading=26,
alignment=TA_CENTER),
"cell": ParagraphStyle("cell", fontName=FONT_BODY, fontSize=8.6, leading=11.5,
alignment=TA_LEFT),
@@ -379,29 +413,24 @@ def build_pdf(order, items, out_path, watermark=None):
sign_line = u"制单:%s\n审批:%s" % (
stamp(order.created_by_name, order.created_at),
stamp(order.approved_by, order.approved_at))
confirm_line = (u"供方确认(盖章):__________________ "
u"需方确认(盖章):__________________")
# One final approval produces one procurement seal. Historical archived
# PDFs are served unchanged by order_pdf and never regenerated here.
if order.approved_by:
seal_cell = SealFlowable(order.our_company or COMPANY_NAME, u"采购专用章")
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"]))
seal = (SealFlowable(order.our_company or COMPANY_NAME, u"采购专用章")
if order.approved_by else None)
party_width = 83 * mm
confirmations = Table(
[[ConfirmationFlowable(u"供方", party_width), "",
ConfirmationFlowable(u"需方", party_width, seal)]],
colWidths=[party_width, 20 * mm, party_width])
confirmations.setStyle(TableStyle([
("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
("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([P(sign_line, styles["sign"]),
Spacer(1, 3 * mm), confirmations]))
return story
doc.build(make_story())
+70
View File
@@ -130,6 +130,58 @@ table.grid tr.row-off .btn { opacity: .8; }
.badge.track-partial { background: #e6f0ff; color: #14509b; }
.badge.track-received, .badge.track-delivered, .badge.track-paid { background: #e8f6ee; color: #12703f; }
.mobile-only { display: none; }
/* Payment / delivery workspace: independent states, task-first rows. */
.tracking-dashboard .card-head { align-items: flex-start; }
.tracking-dashboard .card-head > a { flex-shrink: 0; white-space: nowrap; }
.tracking-dashboard .card-head p { margin: 6px 0 0; }
.tracking-focus .stat { position: relative; border-top: 3px solid #d99a30; padding-right: 28px; }
.tracking-focus .focus-paid_undelivered { border-top-color: #1668dc; }
.tracking-focus .focus-recheck { border-top-color: #a65d26; }
.tracking-focus .stat.selected { background: #eef5ff; border-color: #1668dc; }
.tracking-focus .stat .k { color: #42526a; }
.stat-arrow { position: absolute; right: 12px; top: 24px; color: #697a91; }
.tracking-board-head { display: flex; align-items: center; justify-content: space-between; flex-wrap: wrap; gap: 8px; }
.tracking-board-head h3 { margin: 0; }
.tracking-filter { margin: 0 0 10px; flex-wrap: nowrap; }
.tracking-filter select { width: auto; max-width: 240px; }
.tracking-filter label { white-space: nowrap; }
.status-steps { display: flex; list-style: none; margin: 0; padding: 0; gap: 3px; }
.status-steps li { flex: 1; white-space: nowrap; text-align: center; padding: 6px 3px; border: 1px solid #e2e6ea; border-radius: 4px; font-size: 11px; color: #7b8490; background: #f7f8fa; }
.status-steps .current { font-weight: 600; border-color: currentColor; }
.status-steps .track-unpaid, .status-steps .track-pending { background: #fff4e0; color: #935100; }
.status-steps .track-partial { background: #e6f0ff; color: #14509b; }
.status-steps .track-paid, .status-steps .track-delivered { background: #e8f6ee; color: #12703f; }
.order-progress { min-width: 154px; }
.order-progress .badge, .order-progress .tracking-warning { display: inline-block; margin-bottom: 5px; }
.order-progress.is-stale .current { color: #777; background: #eef0f2; }
.tracking-tasks { display: flex; flex-wrap: wrap; gap: 4px; margin: 5px 0 9px; }
.task-tag { color: #586577; font-size: 12px; line-height: 1.6; }
.task-tag + .task-tag::before { content: '·'; margin-right: 4px; }
.task-prepayment, .task-invoice_due, .task-recheck { color: #935100; font-weight: 600; }
.task-closed { color: #12703f; }
.tracking-actions { display: flex; align-items: center; flex-wrap: wrap; gap: 6px; }
.tracking-actions .btn { white-space: nowrap; }
.tracking-detail-link { font-size: 12px; display: inline-block; padding: 4px 0; }
.supplier-name { margin: 5px 0; overflow-wrap: anywhere; }
.order-identity { max-width: 240px; }
.tracking-table td, .orders-table td { padding-top: 12px !important; padding-bottom: 12px !important; }
.tracking-completed-link { border-top: 1px solid #e2e6ea; margin-top: 12px; padding-top: 12px; font-size: 12px; }
.completed-toggle { display: flex; align-items: center; gap: 4px; margin: 0; color: #555; }
.tracking-panels { display: grid; grid-template-columns: 1fr 1fr; gap: 16px; margin: 12px 0 20px; }
.tracking-panel { padding: 16px; border: 1px solid #dfe5ec; border-radius: 6px; min-width: 0; }
.tracking-panel h3 { margin: 0 0 12px; }
.tracking-panel .status-steps li { font-size: 13px; padding: 9px 3px; }
.follow-note { white-space: pre-wrap; overflow-wrap: anywhere; line-height: 1.7; }
.update-meta { color: #637187; font-size: 12px; margin-bottom: 0; }
#tracking-dialog { width: 540px; max-width: calc(100% - 24px); max-height: calc(100% - 32px); padding: 0; border: 0; border-radius: 10px; box-shadow: 0 16px 60px #0004; }
#tracking-dialog::backdrop { background: #15253b80; }
#tracking-dialog .modal-head { position: sticky; top: 0; background: white; z-index: 1; }
#tracking-dialog h3 { margin: 14px 0; font-size: 15px; }
#tracking-reload { margin-bottom: 12px; }
.editor-order { line-height: 1.8; margin: 0; overflow-wrap: anywhere; }
.tracking-editor-page { max-width: 600px; }
.btn:disabled { opacity: .55; cursor: wait; }
[hidden] { display: none !important; }
@media (max-width: 800px) {
.topbar { height: auto; min-height: 52px; flex-wrap: wrap; gap: 12px; padding: 12px 16px; }
.topbar .brand { width: 100%; }
@@ -143,6 +195,24 @@ table.grid tr.row-off .btn { opacity: .8; }
.stat-row { grid-template-columns: repeat(2, minmax(0, 1fr)); gap: 8px; }
.tracking-summary { grid-template-columns: repeat(2, minmax(0, 1fr)); }
.tracking-forms { grid-template-columns: minmax(0, 1fr); }
.tracking-panels { grid-template-columns: minmax(0, 1fr); }
table.grid.order-cards { display: block; min-width: 0; }
table.order-cards thead { display: none; }
table.order-cards tbody { display: block; }
table.order-cards tr { display: block; border: 1px solid #dfe5ec; border-radius: 8px; padding: 10px 12px; margin-bottom: 12px; background: #fff; }
table.order-cards td { display: flex; gap: 10px; align-items: center; justify-content: space-between; border: 0; padding: 7px 0 !important; text-align: left; min-width: 0; max-width: none; }
table.order-cards td::before { content: attr(data-label); font-size: 12px; color: #637187; flex: 0 0 55px; }
table.order-cards .order-identity { display: block; padding-bottom: 10px !important; border-bottom: 1px solid #edf0f4; margin-bottom: 5px; }
table.order-cards .order-identity::before, table.order-cards .order-next::before { display: none; }
table.order-cards .order-next { display: block; border-top: 1px solid #edf0f4; margin-top: 5px; padding-top: 10px !important; }
table.order-cards .order-progress { flex: 1; min-width: 0; max-width: 320px; }
table.order-cards .tracking-tasks { flex: 1; }
table.order-cards td[data-label="收票 / 待办"] { flex-wrap: wrap; }
table.order-cards .empty-row td { display: block; }
.tracking-actions .btn { display: inline-flex; align-items: center; }
.tracking-filter { width: 100%; }
.tracking-filter select { flex: 1; max-width: none; min-width: 0; }
.tracking-panel { padding: 12px; }
table.grid { min-width: 680px; }
table.approval-flow { min-width: 0; }
.mobile-only { display: block; }
+85
View File
@@ -0,0 +1,85 @@
/* Reuse server forms and existing permissions; keep edits on the current list. */
(function () {
'use strict';
var dialog = document.getElementById('tracking-dialog');
if (!dialog || typeof dialog.showModal !== 'function' || !window.fetch) return;
var content = document.getElementById('tracking-dialog-content');
var errorBox = document.getElementById('tracking-dialog-error');
var reload = document.getElementById('tracking-reload');
var close = dialog.querySelector('[data-tracking-close]');
var editorUrl = '', serial = 0, saving = false, opener;
function showError(message) {
errorBox.textContent = message;
errorBox.hidden = false;
reload.hidden = false;
}
function errorText(html) {
var doc = new DOMParser().parseFromString(html, 'text/html');
var message = doc.querySelector('.wrap .card p, body > p');
return message ? message.textContent : '暂时无法完成操作,请重新读取当前状态后再试。';
}
async function loadEditor() {
var current = ++serial;
errorBox.hidden = reload.hidden = true;
content.textContent = '正在读取最新状态…';
try {
var response = await fetch(editorUrl + '?fragment=1', {cache: 'no-store', credentials: 'same-origin'});
var html = await response.text();
if (current !== serial || !dialog.open) return;
if (response.redirected) { window.location.assign(response.url); return; }
if (!response.ok) { content.textContent = ''; showError(errorText(html)); return; }
content.innerHTML = html;
var field = content.querySelector('select, input:not([type=hidden])');
if (field) field.focus();
} catch (error) {
if (current === serial && dialog.open) { content.textContent = ''; showError('读取失败,请检查连接后重试。'); }
}
}
document.addEventListener('click', function (event) {
var link = event.target.closest('[data-quick-tracking]');
if (!link || event.ctrlKey || event.metaKey || event.shiftKey || event.altKey) return;
event.preventDefault();
opener = link;
editorUrl = link.href;
dialog.showModal();
loadEditor();
});
close.addEventListener('click', function () { if (!saving) dialog.close(); });
dialog.addEventListener('cancel', function (event) { if (saving) event.preventDefault(); });
dialog.addEventListener('close', function () {
++serial;
content.textContent = '';
if (opener) opener.focus();
});
reload.addEventListener('click', loadEditor);
dialog.addEventListener('submit', async function (event) {
var form = event.target;
event.preventDefault();
if (saving) return;
var data = new FormData(form);
var submit = form.querySelector('[type=submit]');
saving = true;
close.disabled = reload.disabled = submit.disabled = true;
errorBox.hidden = reload.hidden = true;
var retry = true;
try {
var response = await fetch(form.action, {method: 'POST', body: data,
credentials: 'same-origin', headers: {'X-Tracking-Modal': '1'}});
if (response.redirected) { window.location.assign(response.url); return; }
if (response.ok && (response.headers.get('Content-Type') || '').indexOf('application/json') >= 0) {
var result = await response.json();
if (result.ok) { window.location.reload(); return; }
}
retry = response.status === 400;
showError(errorText(await response.text()));
} catch (error) {
retry = false;
showError('连接中断,无法确认是否保存成功。请重新读取当前状态,核对后再操作。');
} finally {
saving = false;
close.disabled = reload.disabled = false;
submit.disabled = !retry;
}
});
})();
+11
View File
@@ -36,5 +36,16 @@
{% block body %}{% endblock %}
</div>
<footer class="foot">{{ COMPANY_NAME }} · 采购订单管理系统 v4.0</footer>
{% if session.get('user_id') %}
<dialog id="tracking-dialog" aria-labelledby="tracking-dialog-title">
<div class="modal-head"><span id="tracking-dialog-title">快捷登记</span><button type="button" class="btn sm" data-tracking-close aria-label="关闭快捷登记">关闭</button></div>
<div class="modal-body">
<div id="tracking-dialog-error" class="flash error" role="alert" hidden></div>
<button type="button" class="btn sm" id="tracking-reload" hidden>重新读取当前状态</button>
<div id="tracking-dialog-content" aria-live="polite"></div>
</div>
</dialog>
<script src="{{ url_for('static', filename='tracking.js') }}" defer></script>
{% endif %}
</body>
</html>
+28 -19
View File
@@ -8,29 +8,38 @@
<div class="stat"><div class="v">{{ '%.2f'|format(month_total) }}</div><div class="k">审批通过订单总额(元)</div></div>
</div>
<div class="card tracking-dashboard">
<div class="card-head"><h2>付款与交货概况</h2><a href="{{ url_for('orders', status='approved') }}">查看全部</a></div>
<p class="muted">统计当前可见、审批通过的订单。各类数量可能重叠;部分付款和部分交货按订单状态统计。</p>
<div class="stat-row">
{% for key in ['prepayment','invoice_due','unpaid','paid','undelivered','partial_delivery','paid_undelivered','completed'] %}
<a class="stat tracking-stat" data-tracking="{{ key }}" href="{{ url_for('orders', tracking=key) }}"><div class="v">{{ tracking_counts[key] }}</div><div class="k">{{ TRACKING_FILTERS[key] }}</div></a>
<div class="card tracking-dashboard" id="tracking-board">
{% from 'tracking_macros.html' import progress, invoice, tasks, actions with context %}
<div class="card-head"><div><h2>付款与交货待办</h2><p class="muted">先核对变更,再跟进付款与交货。同一订单可能出现在多个提醒中。</p></div><a href="{{ url_for('orders') }}">全部订单</a></div>
<div class="stat-row tracking-focus">
{% for key in ['prepayment','invoice_due','paid_undelivered','recheck'] %}
<a class="stat tracking-stat focus-{{ key }} {{ 'selected' if tracking_key == key else '' }}" data-tracking="{{ key }}" href="{{ url_for('index', tracking=key) }}#tracking-board"><div class="v">{{ tracking_counts[key] }}</div><div class="k">{{ TRACKING_FILTERS[key] }}</div><span class="stat-arrow" aria-hidden="true">→</span></a>
{% endfor %}
</div>
<div class="toolbar tracking-shortcuts">{% for key in ['partial_payment','delivered','unrecorded','recheck'] %}<a class="btn sm" data-tracking="{{ key }}" href="{{ url_for('orders', tracking=key) }}">{{ TRACKING_FILTERS[key] }}({{ tracking_counts[key] }})</a>{% endfor %}</div>
<h3>最近审批通过的订单</h3>
<table class="grid tracking-table">
<tr><th>订单 / 供应商</th><th class="num">订单金额</th><th>预付款</th><th>收票</th><th>付款</th><th>交货</th><th>操作</th></tr>
{% for o in tracking_rows %}
<tr>
<td><a href="{{ url_for('order_view', oid=o.id) }}#order-tracking">{{ o.order_no }}</a><div class="muted">{{ o.supplier_name }}</div>{% if o.fulfillment_stale or o.payment_stale %}<div class="tracking-warning">变更后待核对</div>{% endif %}</td>
<td class="num">{{ '%.2f'|format(o.total_amount) }}</td><td>{{ '需要' if o.needs_prepayment else '—' }}</td>
<td><span class="badge track-{{ o.invoice_status }}">{{ o.invoice_status_text }}</span></td>
<td><span class="badge track-{{ o.payment_status }}">{{ o.payment_status_text }}</span></td>
<td><span class="badge track-{{ o.delivery_status }}">{{ o.delivery_status_text }}</span></td>
<td><a class="btn sm" href="{{ url_for('order_view', oid=o.id) }}#order-tracking">查看跟进</a></td>
<div class="tracking-board-head">
<div><h3>{{ TRACKING_FILTERS[tracking_key] }} <span class="muted">共 {{ tracking_total }} 单</span></h3><p class="muted">同类优先显示较早订单;付款、交货、收票均完成后默认收起。</p></div>
<form method="get" action="{{ url_for('index') }}#tracking-board" class="toolbar tracking-filter">
<label for="board-filter">查看</label>
<select id="board-filter" name="tracking">{% for key, label in TRACKING_FILTERS.items() %}<option value="{{ key }}" {{ 'selected' if tracking_key == key else '' }}>{{ label }}({{ tracking_counts[key] }})</option>{% endfor %}</select>
<button class="btn sm">筛选</button>
</form>
</div>
<table class="grid tracking-table order-cards">
<thead><tr><th>订单 / 供应商</th><th class="num">金额</th><th>付款进度</th><th>交货进度</th><th>收票</th><th>待办 / 操作</th></tr></thead>
<tbody>{% for o in tracking_rows %}
<tr data-order-id="{{ o.id }}">
<td data-label="订单" class="order-identity"><a href="{{ url_for('order_view', oid=o.id) }}#order-tracking"><b>{{ o.order_no }}</b></a><div class="supplier-name">{{ o.supplier_name }}</div>{% if o.needs_prepayment %}<span class="badge track-partial">需提前付款{{ '(待核对)' if o.fulfillment_stale else '' }}</span>{% endif %}</td>
<td data-label="订单金额" class="num">¥{{ '%.2f'|format(o.total_amount) }}</td>
<td data-label="付款">{{ progress(o, 'payment') }}</td>
<td data-label="交货">{{ progress(o, 'delivery') }}</td>
<td data-label="收票">{{ invoice(o) }}</td>
<td data-label="待办 / 操作" class="order-next">{{ tasks(o) }}{{ actions(o, user) }}</td>
</tr>
{% else %}<tr><td colspan="7" class="center muted">暂无审批通过的订单</td></tr>{% endfor %}
{% else %}<tr class="empty-row"><td colspan="6" class="center muted">当前分类暂无订单。可切换分类查看其他订单。</td></tr>{% endfor %}</tbody>
</table>
{% if tracking_total > tracking_rows|length %}<p class="right"><a href="{{ url_for('orders', tracking=tracking_key) }}">查看全部 {{ tracking_total }} 单 →</a></p>{% endif %}
{% if tracking_key != 'open' %}<p><a href="{{ url_for('index') }}#tracking-board">返回全部待跟进({{ tracking_counts['open'] }})</a></p>{% endif %}
<div class="tracking-completed-link"><a data-tracking="closed" href="{{ url_for('index', tracking='closed') }}#tracking-board">查看已全部完成({{ tracking_counts['closed'] }})</a></div>
</div>
{% if todo %}
+25 -39
View File
@@ -1,44 +1,30 @@
{% from 'tracking_macros.html' import progress, invoice, tasks, form with context %}
{% set recent = latest_tracking(o) %}
<div class="card" id="order-tracking">
<h2>付款与交货跟进</h2>
<div class="tracking-summary">
<div><label>预付款安排</label><b>{{ '需提前付款' if o.needs_prepayment else '正常付款' }}</b></div>
<div><label>收票情况</label><span class="badge track-{{ o.invoice_status }}">{{ o.invoice_status_text }}</span></div>
<div><label>付款情况</label><span class="badge track-{{ o.payment_status }}">{{ o.payment_status_text }}</span>{% if o.payment_date %}<div class="muted">{{ o.payment_date }}</div>{% endif %}</div>
<div><label>交货情况</label><span class="badge track-{{ o.delivery_status }}">{{ o.delivery_status_text }}</span></div>
</div>
{% if o.fulfillment_stale or o.payment_stale %}
<p class="hint">订单内容已变更,以下记录来自变更前:{{ '采购跟进' if o.fulfillment_stale else '' }}{{ '、' if o.fulfillment_stale and o.payment_stale else '' }}{{ '付款情况' if o.payment_stale else '' }}。请对应经办人核对后保存确认。</p>
{% endif %}
<div class="card-head"><h2>付款与交货跟进</h2><span class="badge track-{{ 'partial' if o.needs_prepayment else 'unknown' }}">{{ '需提前付款' if o.needs_prepayment else '正常付款' }}{{ '(待核对)' if o.fulfillment_stale else '' }}</span></div>
{{ tasks(o) }}
{% if o.fulfillment_stale or o.payment_stale %}<p class="hint">订单内容已变更,下面保留变更前记录,请对应经办人核对后保存确认。</p>{% endif %}
{% if o.status != 'approved' %}<p class="muted">审批通过后可登记付款、收票和交货情况。已有跟进记录保留供查询。</p>{% endif %}
{% if o.fulfillment_note %}<p>采购跟进备注:{{ o.fulfillment_note }}</p>{% endif %}
{% if o.payment_note %}<p>付款备注:{{ o.payment_note }}</p>{% endif %}
<div class="tracking-panels">
<section class="tracking-panel">
<h3>付款情况</h3>
{{ progress(o, 'payment') }}
{% if o.payment_date %}<p class="muted">实际付款日期:{{ o.payment_date }}</p>{% endif %}
{% if o.payment_note %}<p class="follow-note">{{ o.payment_note }}</p>{% endif %}
{% set log = recent.get('登记付款状态') %}
<p class="update-meta">{% if log %}最近登记:{{ log.created_at.strftime('%Y-%m-%d %H:%M') }} · {{ log.user_name }}{% else %}尚无付款登记记录{% endif %}</p>
</section>
<section class="tracking-panel">
<h3>交货与收票</h3>
{{ progress(o, 'delivery') }}
<p>收票:{{ invoice(o) }}</p>
{% if o.fulfillment_note %}<p class="follow-note">{{ o.fulfillment_note }}</p>{% endif %}
{% set log = recent.get('更新采购跟进') %}
<p class="update-meta">{% if log %}最近跟进:{{ log.created_at.strftime('%Y-%m-%d %H:%M') }} · {{ log.user_name }}{% else %}尚无采购跟进记录{% endif %}</p>
</section>
</div>
<div class="tracking-forms">
{% if may_follow %}
<form id="follow-up-form" method="post" action="{{ url_for('order_follow_up', oid=o.id) }}">
<h3>制单员跟进</h3>
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<input type="hidden" name="row_version" value="{{ o.row_version }}">
<div class="field"><label for="needs-prepayment">付款安排</label><select id="needs-prepayment" name="needs_prepayment"><option value="0" {{ 'selected' if not o.needs_prepayment else '' }}>正常付款</option><option value="1" {{ 'selected' if o.needs_prepayment else '' }}>需提前付款</option></select></div>
<div class="form-grid two">
<div class="field"><label for="invoice-status">收票情况</label><select id="invoice-status" name="invoice_status">{% for key, label in INVOICE_STATUS.items() %}<option value="{{ key }}" {{ 'selected' if o.invoice_status == key else '' }}>{{ label }}</option>{% endfor %}</select></div>
<div class="field"><label for="delivery-status">交货情况</label><select id="delivery-status" name="delivery_status">{% for key, label in DELIVERY_STATUS.items() %}<option value="{{ key }}" {{ 'selected' if o.delivery_status == key else '' }}>{{ label }}</option>{% endfor %}</select></div>
</div>
<div class="field"><label for="fulfillment-note">跟进备注(选填)</label><textarea id="fulfillment-note" name="fulfillment_note" maxlength="500" placeholder="例如:先付30%预付款,已收到第一批货">{{ o.fulfillment_note }}</textarea></div>
<div class="field"><label for="follow-up-reason">更正原因(回退状态时必填)</label><input id="follow-up-reason" name="reason" maxlength="500"></div>
<button class="btn primary">保存采购跟进</button>
</form>
{% endif %}
{% if may_pay %}
<form id="payment-form" method="post" action="{{ url_for('order_payment', oid=o.id) }}">
<h3>审批员登记付款</h3>
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<input type="hidden" name="row_version" value="{{ o.row_version }}">
<div class="field"><label for="payment-status">付款状态</label><select id="payment-status" name="payment_status">{% for key, label in PAYMENT_STATUS.items() %}<option value="{{ key }}" {{ 'selected' if o.payment_status == key else '' }}>{{ label }}</option>{% endfor %}</select></div>
<div class="field"><label for="payment-date">实际付款日期(部分付款或已付款时必填)</label><input id="payment-date" type="date" name="payment_date" value="{{ o.payment_date }}"></div>
<div class="field"><label for="payment-note">付款备注(选填)</label><textarea id="payment-note" name="payment_note" maxlength="500" placeholder="例如:已付首款,剩余尾款待交货后支付">{{ o.payment_note }}</textarea></div>
<div class="field"><label for="payment-reason">更正原因(回退状态时必填)</label><input id="payment-reason" name="reason" maxlength="500"></div>
<button class="btn primary">保存付款状态</button>
</form>
{% endif %}
{% if may_follow %}{{ form(o, 'follow-up') }}{% endif %}
{% if may_pay %}{{ form(o, 'payment') }}{% endif %}
</div>
</div>
+3 -1
View File
@@ -19,6 +19,8 @@
<a class="btn" href="{{ url_for('orders') }}">返回列表</a>
</div>
{% if o.status == 'approved' %}{% include 'order_tracking.html' %}{% endif %}
<div class="card">
<h2>单据信息</h2>
<p class="mobile-only muted">单据信息和物料明细可左右滑动查看。</p>
@@ -81,7 +83,7 @@
{% if o.note %}<p class="muted">补充说明:{{ o.note }}</p>{% endif %}
</div>
{% include 'order_tracking.html' %}
{% if o.status != 'approved' %}{% include 'order_tracking.html' %}{% endif %}
<div class="card">
<h2>审批流程</h2>
+15 -29
View File
@@ -1,5 +1,6 @@
{% extends "base.html" %}
{% block body %}
{% from 'tracking_macros.html' import progress, invoice, tasks, actions with context %}
<h1 class="page">订单列表</h1>
<div class="card">
<form class="toolbar" method="get">
@@ -23,41 +24,26 @@
<option value="">全部跟进情况</option>
{% for key, label in TRACKING_FILTERS.items() %}<option value="{{ key }}" {{ 'selected' if args.get('tracking') == key else '' }}>{{ label }}</option>{% endfor %}
</select>
<label class="completed-toggle"><input type="checkbox" name="show_completed" value="1" {{ 'checked' if args.get('show_completed') == '1' else '' }}> 显示全部完成订单</label>
<button class="btn primary">查询</button>
<span class="grow"></span>
<a class="btn" href="{{ url_for('export_ledger') }}">导出 Excel 台账</a>
{% if user.role in ['maker','admin'] %}<a class="btn primary" href="{{ url_for('order_new') }}">新建订单</a>{% endif %}
</form>
<table class="grid">
<tr>
<th>单号</th><th>类型</th><th>供应商</th><th>采购日期</th><th class="num">金额</th>
<th>审批状态</th><th>付款与交货</th><th>制单</th><th>审批</th><th class="nowrap">操作</th>
{% if not args.get('tracking') and args.get('show_completed') != '1' %}<p class="muted">已收起付款、交货、收票均完成的订单。勾选“显示全部完成订单”可查看。</p>{% endif %}
<table class="grid orders-table order-cards">
<thead><tr><th>订单 / 供应商</th><th>审批状态</th><th class="num">金额</th><th>付款进度</th><th>交货进度</th><th>收票 / 待办</th><th>操作</th></tr></thead>
<tbody>{% for o in rows %}
<tr data-order-id="{{ o.id }}">
<td data-label="订单" class="order-identity"><a href="{{ url_for('order_view', oid=o.id) }}"><b>{{ o.order_no }}</b></a><div class="supplier-name">{{ o.supplier_name }}</div><div class="muted">{{ o.doc_type_text }} · {{ o.order_date }}</div><div class="muted">制单 {{ o.created_by_name }} · 审批 {{ o.approved_by or '—' }}</div>{% if o.needs_prepayment %}<span class="badge track-partial">需提前付款{{ '(待核对)' if o.fulfillment_stale else '' }}</span>{% endif %}</td>
<td data-label="审批"><span class="badge {{ o.status }}">{{ o.status_text }}</span></td>
<td data-label="订单金额" class="num">¥{{ '%.2f'|format(o.total_amount) }}</td>
<td data-label="付款">{{ progress(o, 'payment') }}</td>
<td data-label="交货">{{ progress(o, 'delivery') }}</td>
<td data-label="收票 / 待办">{{ invoice(o) }}{{ tasks(o) }}</td>
<td data-label="操作" class="order-next">{{ actions(o, user) }}<a class="tracking-detail-link" href="{{ url_for('order_pdf', oid=o.id) }}" target="_blank">下载 / 预览 PDF</a></td>
</tr>
{% for o in rows %}
<tr>
<td class="nowrap">{{ o.order_no }}</td>
<td>{{ o.doc_type_text }}</td>
<td>{{ o.supplier_name }}</td>
<td class="nowrap">{{ o.order_date }}</td>
<td class="num">{{ '%.2f'|format(o.total_amount) }}</td>
<td><span class="badge {{ o.status }}">{{ o.status_text }}</span></td>
<td>
{% if o.needs_prepayment %}<span class="badge track-partial">需提前付款</span>{% endif %}
<span class="badge track-{{ o.invoice_status }}">收票:{{ o.invoice_status_text }}</span>
<span class="badge track-{{ o.payment_status }}">付款:{{ o.payment_status_text }}</span>
<span class="badge track-{{ o.delivery_status }}">交货:{{ o.delivery_status_text }}</span>
{% if o.fulfillment_stale or o.payment_stale %}<div class="tracking-warning">变更后待核对</div>{% endif %}
</td>
<td>{{ o.created_by_name }}</td>
<td>{{ o.approved_by or '-' }}</td>
<td class="nowrap">
<a class="btn sm" href="{{ url_for('order_view', oid=o.id) }}">查看</a>
<a class="btn sm" href="{{ url_for('order_pdf', oid=o.id) }}" target="_blank">PDF</a>
</td>
</tr>
{% else %}
<tr><td colspan="10" class="center muted">没有符合条件的订单</td></tr>
{% endfor %}
{% else %}<tr class="empty-row"><td colspan="7" class="center muted">没有符合条件的订单</td></tr>{% endfor %}</tbody>
</table>
</div>
{% endblock %}
+6 -3
View File
@@ -8,6 +8,7 @@
<div class="form-grid">
<div><label>编号</label><input type="text" name="code" placeholder="S003"></div>
<div><label>名称 *</label><input type="text" name="name" required></div>
<div><label>公司简称(用于 PDF 文件名)</label><input type="text" name="short_name" maxlength="60" placeholder="如:精锐五金;留空使用全名"></div>
<div><label>联系人</label><input type="text" name="contact"></div>
<div><label>电话</label><input type="text" name="phone"></div>
<div><label>付款方式</label><input type="text" name="payment_terms" placeholder="月结30天"></div>
@@ -28,13 +29,14 @@
<tr><th>编号</th><th>名称</th><th>联系人</th><th>电话</th><th>付款方式</th><th>税号</th><th>开户银行</th><th>银行帐号</th><th>操作</th></tr>
{% for s in rows %}
<tr class="{% if not s.active %}row-off{% endif %}">
<td>{{ s.code }}</td><td>{{ s.name }}</td><td>{{ s.contact }}</td><td>{{ s.phone }}</td>
<td>{{ s.code }}</td><td>{{ s.name }}{% if s.short_name %}<div class="hint">简称:{{ s.short_name }}</div>{% endif %}</td><td>{{ s.contact }}</td><td>{{ s.phone }}</td>
<td>{{ s.payment_terms }}</td><td>{{ s.tax_no }}</td><td>{{ s.bank_name }}</td>
<td>{{ s.bank_account }}</td>
<td class="nowrap">
<button type="button" class="btn sm" onclick="openEdit(this)"
data-id="{{ s.id }}" data-code="{{ s.code or '' }}"
data-name="{{ s.name or '' }}" data-contact="{{ s.contact or '' }}"
data-short_name="{{ s.short_name or '' }}"
data-phone="{{ s.phone or '' }}"
data-payment_terms="{{ s.payment_terms or '' }}"
data-tax_no="{{ s.tax_no or '' }}"
@@ -69,7 +71,7 @@
<form method="post" action="{{ url_for('supplier_import_excel') }}" enctype="multipart/form-data">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<div class="modal-body">
<div class="hint">先下载模板按格式填好再导入。<b>名称</b>必填;已存在的按下方方式处理(优先按编号匹配,无编号则按名称)。</div>
<div class="hint">先下载模板按格式填好再导入。<b>名称</b>必填;已存在的按下方方式处理(优先按编号匹配,无编号则按名称)。公司简称用于 PDF 文件名;旧表没有简称列时保留原简称,简称列留空则清除原简称。</div>
<div class="field">
<label>1. 下载模板</label>
<a href="{{ url_for('supplier_import_template') }}">供应商导入模板.xlsx</a>
@@ -107,6 +109,7 @@
<div class="form-grid two">
<div><label>编号</label><input type="text" name="code" id="edit_code"></div>
<div><label>名称 *</label><input type="text" name="name" id="edit_name" required></div>
<div><label>公司简称(用于 PDF 文件名)</label><input type="text" name="short_name" id="edit_short_name" maxlength="60" placeholder="留空使用全名"></div>
<div><label>联系人</label><input type="text" name="contact" id="edit_contact"></div>
<div><label>电话</label><input type="text" name="phone" id="edit_phone"></div>
<div><label>付款方式</label><input type="text" name="payment_terms" id="edit_payment_terms"></div>
@@ -126,7 +129,7 @@
<script>
function openEdit(btn){
const d = btn.dataset;
['id','code','name','contact','phone','payment_terms','tax_no','bank_name','bank_account']
['id','code','name','short_name','contact','phone','payment_terms','tax_no','bank_name','bank_account']
.forEach(k => {
const el = document.getElementById('edit_' + k);
if (el) el.value = d[k] || '';
+6
View File
@@ -0,0 +1,6 @@
{% extends 'base.html' %}
{% block body %}
<h1 class="page">{{ '登记付款' if kind == 'payment' else '登记收票 / 交货' }}</h1>
<div class="card tracking-editor-page">{% include 'tracking_editor_fragment.html' %}</div>
<a class="btn" href="{{ url_for('order_view', oid=o.id) }}#order-tracking">返回订单</a>
{% endblock %}
@@ -0,0 +1,5 @@
{% from 'tracking_macros.html' import form with context %}
<div data-tracking-editor>
<p class="editor-order"><b>{{ o.order_no }}</b><br>{{ o.supplier_name }} · ¥{{ '%.2f'|format(o.total_amount) }}</p>
{{ form(o, kind) }}
</div>
+53
View File
@@ -0,0 +1,53 @@
{% macro progress(o, kind) -%}
{% set payment = kind == 'payment' %}
{% set state = o.payment_status if payment else o.delivery_status %}
{% set stale = o.payment_stale if payment else o.fulfillment_stale %}
{% set steps = [('unpaid', '未付款'), ('partial', '部分付款'), ('paid', '全部付款')] if payment else [('pending', '未交货'), ('partial', '部分交货'), ('delivered', '全部交货')] %}
<div class="order-progress {{ 'is-stale' if stale else '' }}" data-progress="{{ kind }}">
{% if stale %}<span class="tracking-warning">变更前记录 · 待核对</span>{% elif state == 'unknown' %}<span class="badge track-unknown">未登记</span>{% endif %}
<ol class="status-steps" aria-label="{{ '付款' if payment else '交货' }}状态{{ '(变更前,待核对)' if stale else '' }}">
{% for value, label in steps %}<li class="{{ 'current track-' ~ state if state == value else '' }}" {% if state == value %}aria-current="step"{% endif %}>{{ label }}</li>{% endfor %}
</ol>
</div>
{%- endmacro %}
{% macro invoice(o) -%}
<span class="badge track-{{ 'unknown' if o.fulfillment_stale else o.invoice_status }}">{{ o.invoice_status_text }}{{ '(待核对)' if o.fulfillment_stale else '' }}</span>
{%- endmacro %}
{% macro tasks(o) -%}
<div class="tracking-tasks">{% for key, text in tracking_tasks(o) %}<span class="task-tag task-{{ key }}">{{ text }}</span>{% endfor %}</div>
{%- endmacro %}
{% macro actions(o, user) -%}
<div class="tracking-actions">
{% if can_follow(o, user) %}<a class="btn sm" data-quick-tracking href="{{ url_for('tracking_editor', oid=o.id, kind='follow-up') }}">登记收票 / 交货</a>{% endif %}
{% if can_record_payment(o, user) %}<a class="btn sm primary" data-quick-tracking href="{{ url_for('tracking_editor', oid=o.id, kind='payment') }}">登记付款</a>{% endif %}
<a class="tracking-detail-link" href="{{ url_for('order_view', oid=o.id) }}#order-tracking">详情与记录</a>
</div>
{%- endmacro %}
{% macro form(o, kind) -%}
{% set payment = kind == 'payment' %}
<form id="{{ 'payment-form' if payment else 'follow-up-form' }}" method="post" action="{{ url_for('order_payment' if payment else 'order_follow_up', oid=o.id) }}">
<h3>{{ '审批员登记付款' if payment else '制单员跟进' }}</h3>
{% if (payment and o.payment_stale) or (not payment and o.fulfillment_stale) %}<p class="hint">订单内容已变更,请核对以下变更前记录。保存即确认本次订单的跟进情况。</p>{% endif %}
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<input type="hidden" name="row_version" value="{{ o.row_version }}">
{% if payment %}
<div class="field"><label for="payment-status">付款状态</label><select id="payment-status" name="payment_status">{% for key, label in PAYMENT_STATUS.items() %}<option value="{{ key }}" {{ 'selected' if o.payment_status == key else '' }}>{{ label }}</option>{% endfor %}</select></div>
<div class="field"><label for="payment-date">实际付款日期(部分付款或已付款时必填)</label><input id="payment-date" type="date" name="payment_date" value="{{ o.payment_date }}"></div>
<div class="field"><label for="payment-note">付款备注(选填)</label><textarea id="payment-note" name="payment_note" maxlength="500" placeholder="例如:已付首款,剩余尾款待交货后支付">{{ o.payment_note }}</textarea></div>
<div class="field"><label for="payment-reason">更正原因(回退状态时必填)</label><input id="payment-reason" name="reason" maxlength="500"></div>
{% else %}
<div class="field"><label for="needs-prepayment">付款安排</label><select id="needs-prepayment" name="needs_prepayment"><option value="0" {{ 'selected' if not o.needs_prepayment else '' }}>正常付款</option><option value="1" {{ 'selected' if o.needs_prepayment else '' }}>需提前付款</option></select></div>
<div class="form-grid two">
<div class="field"><label for="invoice-status">收票情况</label><select id="invoice-status" name="invoice_status">{% for key, label in INVOICE_STATUS.items() %}<option value="{{ key }}" {{ 'selected' if o.invoice_status == key else '' }}>{{ label }}</option>{% endfor %}</select></div>
<div class="field"><label for="delivery-status">交货情况</label><select id="delivery-status" name="delivery_status">{% for key, label in DELIVERY_STATUS.items() %}<option value="{{ key }}" {{ 'selected' if o.delivery_status == key else '' }}>{{ label }}</option>{% endfor %}</select></div>
</div>
<div class="field"><label for="fulfillment-note">跟进备注(选填)</label><textarea id="fulfillment-note" name="fulfillment_note" maxlength="500" placeholder="例如:首批货已到,剩余预计周五交齐">{{ o.fulfillment_note }}</textarea></div>
<div class="field"><label for="follow-up-reason">更正原因(回退状态时必填)</label><input id="follow-up-reason" name="reason" maxlength="500"></div>
{% endif %}
<button class="btn primary" type="submit">{{ '保存付款状态' if payment else '保存采购跟进' }}</button>
</form>
{%- endmacro %}
+39 -1
View File
@@ -7,6 +7,44 @@ import pytest
SCHEMA = Path(__file__).parent / 'fixtures' / 'legacy_schema.sql'
def test_v4_supplier_alias_upgrade_preserves_orders_and_creates_one_backup(tmp_path):
from models import Order, OrderVersion, Supplier, init_db
path = tmp_path / 'po.db'
engine, factory = init_db(str(path))
archived = b'%PDF-1.4 immutable before alias update'
with factory() as db:
db.add(Supplier(name='旧供应商', code='S01', contact='原联系人'))
db.add(Order(id=1, order_no='V4-ORDER', status='approved', revision=2,
payment_status='partial', invoice_status='received', delivery_status='delivered'))
db.flush()
db.add(OrderVersion(order_id=1, revision=2, content_json='{}', pdf_data=archived,
pdf_sha256='existing-digest', template_version='po-v4'))
db.commit()
engine.dispose()
with sqlite3.connect(path) as con:
con.execute('ALTER TABLE suppliers DROP COLUMN short_name')
con.execute('PRAGMA user_version=4')
old_orders = con.execute('SELECT * FROM orders').fetchall()
old_archives = con.execute('SELECT * FROM order_versions').fetchall()
engine, factory = init_db(str(path))
with factory() as db:
supplier = db.query(Supplier).one()
assert (supplier.name, supplier.code, supplier.contact, supplier.short_name) == ('旧供应商', 'S01', '原联系人', '')
engine.dispose()
with sqlite3.connect(path) as con:
assert con.execute('PRAGMA user_version').fetchone()[0] == 5
assert con.execute('SELECT * FROM orders').fetchall() == old_orders
assert con.execute('SELECT * FROM order_versions').fetchall() == old_archives
backups = list(tmp_path.glob('*.bak'))
assert len(backups) == 1
with sqlite3.connect(backups[0]) as con:
assert con.execute('PRAGMA user_version').fetchone()[0] == 4
assert 'short_name' not in {row[1] for row in con.execute('PRAGMA table_info(suppliers)')}
engine, _ = init_db(str(path))
engine.dispose()
assert len(list(tmp_path.glob('*.bak'))) == 1
def test_v3_tracking_upgrade_preserves_existing_state_and_archives(tmp_path):
import hashlib
from models import Order, OrderVersion, User, init_db
@@ -48,7 +86,7 @@ def test_v3_tracking_upgrade_preserves_existing_state_and_archives(tmp_path):
assert con.execute('PRAGMA user_version').fetchone()[0] == 3
assert 'payment_status' not in {row[1] for row in con.execute('PRAGMA table_info(orders)')}
with sqlite3.connect(path) as con:
assert con.execute('PRAGMA user_version').fetchone()[0] == 4
assert con.execute('PRAGMA user_version').fetchone()[0] == 5
assert con.execute('PRAGMA integrity_check').fetchone()[0] == 'ok'
engine, _ = init_db(str(path))
engine.dispose()
+98
View File
@@ -0,0 +1,98 @@
import io
import re
import openpyxl
import pytest
from werkzeug.http import parse_options_header
from conftest import post
from test_workflow import action, new, order, save
def download_name(response):
return parse_options_header(response.headers['Content-Disposition'])[1]['filename']
@pytest.mark.parametrize('kind,title', [('purchase', '采购单'), ('outsourcing', '委外生产订单')])
def test_supplier_alias_updates_filename_without_changing_archive(app, clients, kind, title):
company = '苏州精锐五金制品有限公司'
assert post(clients['maker'], '/suppliers', {'name': company, 'short_name': '精锐五金'}).status_code == 302
with app.Session() as db:
sid = db.query(app.Supplier).one().id
oid = new(app, clients['maker'], kind)
assert save(app, clients['maker'], oid, supplier_name=company, intent='submit').status_code == 302
assert action(app, clients['approver'], oid, 'approve', stamped='1').status_code == 302
with app.Session() as db:
number = db.get(app.Order, oid).order_no
path = '/orders/%s/pdf' % oid
archive = clients['maker'].get(path).data
original_version = order(app, oid)['version']
for suffix, disposition in [('', 'inline'), ('?dl=1', 'attachment')]:
response = clients['maker'].get(path + suffix)
assert response.headers['Content-Disposition'].startswith(disposition + ';')
assert download_name(response) == '%s_%s_精锐五金.pdf' % (title, number)
assert response.data == archive
assert post(clients['maker'], '/suppliers', {'id': sid, 'name': company, 'short_name': '精锐'}).status_code == 302
response = clients['maker'].get(path + '?dl=1')
assert download_name(response) == '%s_%s_精锐.pdf' % (title, number)
assert response.data == archive and order(app, oid)['version'] == original_version
assert clients['other'].get(path + '?dl=1').status_code == 404
# A cleared alias or removed master-data entry still yields an identifiable name.
assert post(clients['maker'], '/suppliers', {'id': sid, 'name': company, 'short_name': ''}).status_code == 302
assert download_name(clients['maker'].get(path)) == '%s_%s_%s.pdf' % (title, number, company)
assert post(clients['admin'], '/suppliers/%s/delete' % sid).status_code == 302
response = clients['maker'].get(path)
assert download_name(response) == '%s_%s_%s.pdf' % (title, number, company)
assert response.data == archive
def test_pdf_filename_handles_unsafe_characters_and_ambiguous_suppliers(app, clients):
company = '测试同名公司'
with app.Session() as db:
db.add(app.Supplier(name=company, tax_no='ONE', short_name='精锐/五金:分部?\r\n<二>'))
db.add(app.Supplier(name=company, tax_no='TWO', short_name='另一家'))
db.commit()
oid = new(app, clients['maker'])
assert save(app, clients['maker'], oid, supplier_name=company, supplier_tax_no='ONE').status_code == 302
response = clients['maker'].get('/orders/%s/pdf?dl=1' % oid)
filename = download_name(response)
assert response.status_code == 200
assert '精锐_五金' in filename and '分部' in filename
assert not re.search(r'[<>:"/\\|?*\x00-\x1f]', filename)
assert len(filename.encode('utf-16-le')) // 2 <= 255
# Without identifying tax information, do not pick an arbitrary company's alias.
assert save(app, clients['maker'], oid, supplier_name=company, supplier_tax_no='').status_code == 302
assert download_name(clients['maker'].get('/orders/%s/pdf' % oid)).endswith('_测试同名公司.pdf')
assert save(app, clients['maker'], oid, supplier_name='公司' * 90).status_code == 302
assert len(download_name(clients['maker'].get('/orders/%s/pdf' % oid)).encode('utf-16-le')) // 2 <= 255
def test_supplier_import_alias_headers_legacy_preservation_and_blank_clear(app, clients):
company = '苏州测试有限公司'
def upload(heads, row):
workbook = openpyxl.Workbook()
workbook.active.append(heads)
workbook.active.append(row)
stream = io.BytesIO()
workbook.save(stream)
stream.seek(0)
response = post(clients['maker'], '/suppliers/import_excel', {'file': (stream, '供应商.xlsx')})
assert response.status_code == 302
upload(['公司简称', '供应商名称', '供应商编号'], ['苏测', company, 'S001'])
with app.Session() as db:
supplier = db.query(app.Supplier).one()
assert (supplier.code, supplier.name, supplier.short_name) == ('S001', company, '苏测')
upload(['编号', '名称', '联系人'], ['S001', company, '新联系人'])
with app.Session() as db:
supplier = db.query(app.Supplier).one()
assert (supplier.short_name, supplier.contact) == ('苏测', '新联系人')
upload(['编号', '名称', '公司简称'], ['S001', company, ''])
with app.Session() as db:
assert db.query(app.Supplier).one().short_name == ''
template = openpyxl.load_workbook(io.BytesIO(clients['maker'].get('/suppliers/import_template.xlsx').data))
assert template.active.cell(1, 9).value == '公司简称'
assert template.active.cell(2, 9).value == '精锐五金'
template.close()
assert post(clients['maker'], '/suppliers', {'name': '超长简称测试', 'short_name': '长' * 61}).status_code == 400
+61 -1
View File
@@ -56,7 +56,7 @@ def test_follow_payment_dashboard_and_archived_pdf(app, clients):
assert not contains(clients['approver'], 'paid_undelivered', number)
assert clients['maker'].get('/orders/%s/pdf' % oid).data == pdf
dashboard = clients['admin'].get('/').get_data(as_text=True)
assert re.search(r'data-tracking="completed"[^>]*><div class="v">1</div>', dashboard)
assert re.search(r'data-tracking="closed"[^>]*>查看已全部完成(1)', dashboard)
other_dashboard = clients['other'].get('/').get_data(as_text=True)
assert number not in other_dashboard
with app.Session() as db:
@@ -160,3 +160,63 @@ def test_admin_can_follow_and_pay_and_invalid_filter_is_rejected(app, clients):
assert follow(app, clients, oid, actor='admin').status_code == 302
assert pay(app, clients, oid, actor='admin').status_code == 302
assert clients['admin'].get('/orders?tracking=invalid').status_code == 400
def test_action_queue_keeps_old_open_orders_and_unreceived_invoices(app, clients):
with app.Session() as db:
maker = db.query(app.User).filter_by(username='maker').one()
for i in range(16):
db.add(app.Order(order_no='QUEUE-%02d' % i, created_by=maker.id,
supplier_name='待办测试', status='approved', revision=1,
payment_status='paid', delivery_status='delivered',
invoice_status='received', fulfillment_revision=1, payment_revision=1))
db.flush()
rows = db.query(app.Order).order_by(app.Order.id).all()
rows[0].payment_status = 'unpaid'
rows[0].needs_prepayment = True
rows[1].invoice_status = 'pending'
rows[2].payment_revision = 0
rows[3].payment_status = 'unknown'
db.commit()
html = clients['maker'].get('/').get_data(as_text=True)
board = html.split('<table class="grid tracking-table order-cards">')[1].split('</table>')[0]
assert all('QUEUE-%02d' % i in board for i in range(4))
assert 'QUEUE-15' not in board
assert board.index('QUEUE-02') < board.index('QUEUE-00')
assert '跟进收票' in board and '变更前记录 · 待核对' in board
assert 'QUEUE-15' not in clients['maker'].get('/orders').get_data(as_text=True)
assert 'QUEUE-15' in clients['maker'].get('/orders?show_completed=1').get_data(as_text=True)
assert 'QUEUE-15' in clients['maker'].get('/orders?tracking=closed').get_data(as_text=True)
assert 'QUEUE-00' not in clients['other'].get('/').get_data(as_text=True)
assert clients['admin'].get('/?tracking=invalid').status_code == 400
@pytest.mark.parametrize('actor,kind,status', [
('maker', 'payment', 403), ('approver', 'follow-up', 403),
('other', 'follow-up', 404), ('maker', 'follow-up', 200),
('approver', 'payment', 200), ('admin', 'follow-up', 200),
])
def test_quick_editor_checks_permissions(app, clients, actor, kind, status):
oid = approved(app, clients)
for suffix in ('', '?fragment=1'):
assert clients[actor].get('/orders/%s/tracking/%s%s' % (oid, kind, suffix)).status_code == status
def test_quick_payment_preserves_concurrency_archive_and_latest_actor(app, clients):
oid = approved(app, clients)
pdf = clients['maker'].get('/orders/%s/pdf' % oid).data
version = order(app, oid)['version']
values = {'row_version': str(version), 'payment_status': 'partial',
'payment_date': date.today().isoformat(), 'payment_note': '已付首款,尾款待交齐'}
response = post(clients['approver'], '/orders/%s/payment' % oid, values,
headers={'X-Tracking-Modal': '1'})
assert response.status_code == 200 and response.json == {'ok': True}
stale = post(clients['approver'], '/orders/%s/payment' % oid, values,
headers={'X-Tracking-Modal': '1'})
assert stale.status_code == 409
assert follow(app, clients, oid).status_code == 302
html = clients['maker'].get('/orders/%s/view' % oid).get_data(as_text=True)
assert '最近登记:' in html and '最近跟进:' in html
assert '已付首款,尾款待交齐' in html and '已收第一批货' in html
assert html.index('付款与交货跟进') < html.index('单据信息')
assert clients['maker'].get('/orders/%s/pdf' % oid).data == pdf
+3 -2
View File
@@ -78,14 +78,15 @@ def test_full_flow_and_frozen_pdf(app, clients, kind, n):
assert clients['maker'].get('/orders/%s/pdf' % oid).data == response.data
import pymupdf
with pymupdf.open(stream=response.data, filetype='pdf') as doc:
content = ''.join(p.get_text() for p in doc)
# Older PDF readers may split adjacent Chinese glyphs into lines.
content = re.sub(r'\s+', '', ''.join(p.get_text() for p in doc))
assert content.count('采购专用章') == 1
assert '审核专用章' not in content and '批准专用章' not in content
with app.Session() as db:
v = db.query(OrderVersion).filter_by(order_id=oid, revision=1).one()
assert v.pdf_sha256 == hashlib.sha256(response.data).hexdigest()
assert len(json.loads(v.content_json)['items']) == n
assert v.template_version == 'po-v3'
assert v.template_version == 'po-v5'
assert save(app, clients['maker'], oid).status_code == 403
assert clients['maker'].get('/orders/%s/edit' % oid).status_code == 302
assert clients['maker'].get('/orders/%s/versions/1' % oid).status_code == 200
+20 -4
View File
@@ -70,6 +70,18 @@ def run(executable=None):
admin.get_by_role('button',name='保存用户').click()
admin.wait_for_url(base + '/users')
admin.screenshot(path=str(RUN / 'users.png'),full_page=True)
admin.goto(base + '/suppliers')
supplier_form = admin.locator('.card').first.locator('form')
supplier_form.locator('[name=name]').fill('测试供应商')
supplier_form.locator('[name=short_name]').fill('测试简称')
supplier_form.get_by_role('button', name='新增', exact=True).click()
admin.wait_for_url(base + '/suppliers')
admin.get_by_role('button', name='编辑', exact=True).click()
assert admin.locator('#edit_short_name').input_value() == '测试简称'
admin.locator('#edit_short_name').fill('测试供应商简称')
admin.locator('#editModal').get_by_role('button', name='保存', exact=True).click()
admin.wait_for_url(base + '/suppliers')
admin.screenshot(path=str(RUN / 'supplier-short-name.png'), full_page=True)
def sign_in(username, destination):
tab=page(mobile=username != 'maker'); tab.goto(destination)
assert '/login' in tab.url
@@ -148,6 +160,9 @@ def run(executable=None):
approver.screenshot(path=str(RUN / (kind+'-approved.png')),full_page=True)
response=approver.request.get(view_url.replace('/view','/pdf'))
assert response.status==200
from werkzeug.http import parse_options_header
filename = parse_options_header(response.headers['content-disposition'])[1]['filename']
assert filename.endswith('_测试供应商简称.pdf')
data=response.body();(RUN/(kind+'.pdf')).write_bytes(data)
archives.append((view_url.replace('/view','/pdf'), hashlib.sha256(data).hexdigest()))
document=pymupdf.open(stream=data,filetype='pdf')
@@ -188,13 +203,14 @@ def run(executable=None):
approver.screenshot(path=str(RUN / (kind+'-mobile-tracking.png')))
assert hashlib.sha256(approver.request.get(view_url.replace('/view','/pdf')).body()).hexdigest() == archives[-1][1]
admin.goto(base + '/')
for key in ('prepayment', 'invoice_due', 'partial_delivery', 'paid_undelivered'):
for key in ('prepayment', 'invoice_due', 'paid_undelivered'):
assert admin.locator('[data-tracking="'+key+'"] .v').inner_text() == '1', key
assert admin.locator('#board-filter option[value="partial_delivery"]').inner_text() == '部分交货(1)'
admin.screenshot(path=str(RUN / 'tracking-dashboard.png'), full_page=True)
admin.locator('[data-tracking="paid_undelivered"]').click()
admin.wait_for_load_state('networkidle')
assert admin.locator('table.grid tr').count() == 2
assert 'WO' in admin.locator('table.grid tr').nth(1).inner_text()
assert admin.locator('.tracking-table tbody tr').count() == 1
assert 'WO' in admin.locator('.tracking-table tbody tr').first.inner_text()
admin.screenshot(path=str(RUN / 'tracking-filter.png'), full_page=True)
for width in (320, 390, 768):
approver.set_viewport_size({'width':width, 'height':844})
@@ -228,7 +244,7 @@ def run(executable=None):
assert hashlib.sha256(response.body()).hexdigest() == expected
with sqlite3.connect(RUN / 'instance' / 'po.db') as db:
assert db.execute('PRAGMA integrity_check').fetchone()[0] == 'ok'
assert db.execute('PRAGMA user_version').fetchone()[0] == 4
assert db.execute('PRAGMA user_version').fetchone()[0] == 5
assert db.execute('SELECT count(*) FROM orders WHERE status = ?', ('approved',)).fetchone()[0] == 2
assert db.execute('SELECT count(*) FROM notifications').fetchone()[0] == 0
assert db.execute("SELECT count(*) FROM orders WHERE payment_status='paid' AND delivery_status='pending'").fetchone()[0] == 1
+140
View File
@@ -0,0 +1,140 @@
"""Quick tracking dialogs and responsive cards, using isolated sample orders."""
import json
import os
import sys
import tempfile
import threading
from datetime import date, datetime
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
RUN = Path(tempfile.mkdtemp(prefix='tracking-ui-', dir=ROOT/'tools/out'))
os.environ['PO_DATA_DIR'] = str(RUN)
sys.path.insert(0, str(ROOT/'po_app'))
from app import app, Session
from models import ApprovalLog, Order, User
from playwright.sync_api import sync_playwright, expect
from waitress.server import create_server
from werkzeug.security import generate_password_hash
PASSWORD = 'Isolated-Tracking-2026'
with Session() as db:
users = {}
for name, role in [('采购员', 'maker'), ('审批员', 'approver'), ('管理员', 'admin')]:
user = User(username=role, name=name, role=role, active=True,
must_change_password=False, session_version=1,
password_hash=generate_password_hash(PASSWORD))
db.add(user)
db.flush()
users[role] = user.id
samples = [
('博远精密(测试)', 'unpaid', 'pending', 'pending', True),
('华星电子(测试)', 'partial', 'delivered', 'received', False),
('恒达机电(测试)', 'paid', 'partial', 'received', False),
('新力材料(测试)', 'paid', 'delivered', 'received', False),
('中科配件(测试)', 'unknown', 'unknown', 'unknown', False),
('宏光模具(测试)', 'paid', 'delivered', 'pending', False),
]
for i, (supplier, payment, delivery, invoice, prepay) in enumerate(samples, 1):
order = Order(order_no='PO20260917-%03d' % i, supplier_name=supplier,
created_by=users['maker'], created_by_name='采购员',
status='approved', order_date='2026-09-17', total_amount=1200*i,
revision=1, approved_by='审批员', approved_at=datetime.now(),
payment_status=payment, delivery_status=delivery, invoice_status=invoice,
needs_prepayment=prepay, fulfillment_revision=1, payment_revision=1)
if i == 4:
order.fulfillment_revision = 0
db.add(order)
db.commit()
server = create_server(app, host='127.0.0.1', port=0)
threading.Thread(target=server.run, daemon=True).start()
base = 'http://127.0.0.1:%s' % server.effective_port
errors = []
try:
with sync_playwright() as p:
browser = p.chromium.launch(channel='msedge', headless=True)
def login(role):
page = browser.new_page(viewport={'width': 1440, 'height': 1100})
page.on('pageerror', lambda error: errors.append(str(error)))
page.goto(base+'/login')
page.locator('[name=username]').fill(role)
page.locator('[name=password]').fill(PASSWORD)
page.get_by_role('button', name='登 录').click()
page.wait_for_url(base+'/')
return page
maker, approver, admin = [login(role) for role in ('maker', 'approver', 'admin')]
admin.screenshot(path=str(RUN/'desktop.png'), full_page=True)
assert maker.locator('[data-quick-tracking][href$="/payment"]').count() == 0
assert approver.locator('[data-quick-tracking][href$="/follow-up"]').count() == 0
assert admin.locator('.tracking-table tbody tr').first.get_attribute('data-order-id') == '4'
for width in (320, 390, 768, 1024):
approver.set_viewport_size({'width': width, 'height': 844})
for path in ('/', '/orders', '/orders/2/view'):
approver.goto(base+path)
assert approver.evaluate('document.documentElement.scrollWidth <= innerWidth'), (width, path)
if width <= 768 and path != '/orders/2/view':
assert approver.locator('table.order-cards').evaluate('(el) => el.scrollWidth <= el.clientWidth + 1'), (width, path)
approver.set_viewport_size({'width': 390, 'height': 844})
approver.goto(base+'/')
approver.screenshot(path=str(RUN/'mobile.png'), full_page=True)
# Inline validation retains user input; a stale form cannot overwrite a newer record.
approver.locator('tr[data-order-id="2"] [data-quick-tracking]').click()
dialog = approver.locator('#tracking-dialog')
expect(dialog.locator('[name=payment_status]')).to_have_value('partial')
dialog.locator('[name=payment_status]').select_option('unpaid')
dialog.get_by_role('button', name='保存付款状态').click()
expect(dialog.locator('#tracking-dialog-error')).to_contain_text('更正原因')
expect(dialog.locator('[name=payment_status]')).to_have_value('unpaid')
dialog.locator('[name=payment_status]').select_option('paid')
dialog.locator('[name=payment_date]').fill(date.today().isoformat())
dialog.locator('[name=payment_note]').fill('款项已付清')
# Another authorized user changes the same order while this dialog is open.
admin.locator('tr[data-order-id="2"] a[href$="/payment"]').click()
ad = admin.locator('#tracking-dialog')
ad.locator('[name=payment_date]').fill(date.today().isoformat())
ad.locator('[name=payment_note]').fill('另一位经办人已核对首款')
ad.get_by_role('button', name='保存付款状态').click()
expect(ad).not_to_be_visible()
dialog.get_by_role('button', name='保存付款状态').click()
expect(dialog.locator('#tracking-dialog-error')).to_contain_text('已经更新')
expect(dialog.get_by_role('button', name='保存付款状态')).to_be_disabled()
dialog.locator('#tracking-reload').click()
expect(dialog.locator('[name=payment_note]')).to_have_value('另一位经办人已核对首款')
dialog.locator('[name=payment_status]').select_option('paid')
dialog.locator('[name=payment_note]').fill('款项已付清')
approver.screenshot(path=str(RUN/'payment-dialog-mobile.png'))
dialog.get_by_role('button', name='保存付款状态').click()
expect(dialog).not_to_be_visible()
expect(approver.locator('.tracking-table tr[data-order-id="2"]')).to_have_count(0)
assert approver.url == base+'/'
# Maker updates delivery and invoice directly from the order list.
maker.goto(base+'/orders?kw=001')
maker.locator('a[data-quick-tracking]').click()
md = maker.locator('#tracking-dialog')
md.locator('[name=invoice_status]').select_option('received')
md.locator('[name=delivery_status]').select_option('partial')
md.locator('[name=fulfillment_note]').fill('首批货已到,剩余预计周五交齐。')
md.get_by_role('button', name='保存采购跟进').click()
expect(md).not_to_be_visible()
assert maker.url == base+'/orders?kw=001'
expect(maker.locator('[data-progress=delivery] [aria-current=step]')).to_have_text('部分交货')
maker.goto(base+'/orders/1/view')
expect(maker.locator('#order-tracking')).to_contain_text('最近跟进:')
expect(maker.locator('#order-tracking')).to_contain_text('首批货已到,剩余预计周五交齐。')
maker.locator('#order-tracking').screenshot(path=str(RUN/'detail.png'))
# A standard page remains usable when JavaScript is unavailable.
fallback = browser.new_context(java_script_enabled=False)
fallback.add_cookies(maker.context.cookies())
tab = fallback.new_page()
tab.goto(base+'/orders/1/tracking/follow-up')
assert tab.locator('#follow-up-form').count() == 1
assert not errors, errors
browser.close()
with Session() as db:
assert db.get(Order, 2).payment_status == 'paid'
assert db.get(Order, 1).delivery_status == 'partial'
assert db.query(ApprovalLog).filter_by(user_name='采购员', action='更新采购跟进').count() == 1
print(json.dumps({'result': 'PASS', 'output': str(RUN), 'js_errors': errors}, ensure_ascii=False))
finally:
server.close()
+3
View File
@@ -10,6 +10,9 @@
4. 在运行系统的电脑上读取 `po_app/instance/setup.token`,把初始化口令填入页面,设置管理员账号和密码。不要发送该文件给无关人员。
5. 管理员在“用户”页新增制单员、审批员,两个环节分配给不同人员。新增账号首次登录必须修改初始密码。
6. 录入真实供应商和物料,再开始制单。新数据库不再预置演示账号、演示供应商或假银行信息。
- 在“供应商管理”填写“公司简称”,用于 PDF 导出文件名,例如 `采购单_PO202609170001_精锐五金.pdf`。未填简称时使用订单上的供应商全名;按全名匹配供应商资料,同名有歧义时结合税号匹配,仍有歧义则使用全名。
- 简称只用于下载命名,PDF 正文仍显示全名;已审批订单重新下载即可使用新简称,归档内容保持不变。文件名中的 Windows 不允许字符会被替换。
- Excel 供应商模板末列支持“公司简称”。旧表没有该列时保留原简称;该列留空则清除原简称。此功能使用数据库结构版本 5,旧数据库启动时先备份再增量升级。
系统使用 Waitress 提供服务。源码和 EXE 均支持 `PO_DATA_DIR` 指定数据目录;未设置时分别使用 `po_app` 目录和 EXE 所在目录。`PO_HOST` 默认 `0.0.0.0`;端口已占用时启动程序会在 8120 之前尝试后续端口,以控制台打印地址为准。