wangbeihong/inventory
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 添加 AI 重复物料巡检功能，支持生成人工复核清单和导出 CSV

Browse Source
This commit is contained in:
wangbeihong
2026-03-12 15:08:52 +08:00
parent ef0af75193
commit 168f5fe49c
4 changed files with 518 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
README.md
View File

@@ -359,11 +359,11 @@ cp /www/wwwroot/inventory/data/inventory.db /www/backup/inventory_$(date +%F).db
### 第二阶段：提升库存数据质量
- [ ] AI 重复物料巡检
- [ ] 定期扫描库存，找出“疑似同料号”“疑似同参数”“疑似同立创编号”的记录
- [ ] 输出“疑似重复物料清单”，由人工决定是否合并
- [ ] 给出重复原因说明，例如名称近似、规格一致、备注含相同 LCSC 编号
- [ ] 增加“建议统一名称/规格写法”的辅助提示
- [x] AI 重复物料巡检
- [x] 定期扫描库存，找出“疑似同料号”“疑似同参数”“疑似同立创编号”的记录
- [x] 输出“疑似重复物料清单”，由人工决定是否合并
- [x] 给出重复原因说明，例如名称近似、规格一致、备注含相同 LCSC 编号
- [x] 增加“建议统一名称/规格写法”的辅助提示
### 第三阶段：升级补货能力

326
app.py
View File

@@ -1491,6 +1491,184 @@ def _build_restock_payload(*, limit: int = 20, threshold: int = LOW_STOCK_THRESH
}
def _normalize_part_no_key(part_no: str) -> str:
return (part_no or "").strip().upper()
def _pick_standard_text(values: list[str]) -> str:
"""从多个文本候选中挑选建议标准写法。
中文说明：优先选择出现次数最多的写法；若次数相同，选择长度更长的写法，
这样一般能保留更多有效信息。
"""
bucket = {}
for value in values:
text = (value or "").strip()
if not text:
continue
bucket[text] = bucket.get(text, 0) + 1
if not bucket:
return ""
ordered = sorted(bucket.items(), key=lambda item: (-item[1], -len(item[0]), item[0]))
return ordered[0][0]
def _build_duplicate_member(component: Component, box_by_id: dict[int, Box]) -> dict:
box = box_by_id.get(component.box_id)
lcsc_code = _extract_lcsc_code_from_text(component.note or "")
if not lcsc_code:
lcsc_code = _extract_lcsc_code_from_text(component.part_no or "")
return {
"component_id": component.id,
"part_no": component.part_no or "",
"name": component.name or "",
"specification": component.specification or "",
"quantity": int(component.quantity or 0),
"lcsc_code": lcsc_code,
"box_id": component.box_id,
"box_name": box.name if box else f"盒 {component.box_id}",
"slot_code": slot_code_for_box(box, component.slot_index) if box else str(component.slot_index),
"edit_url": url_for("edit_component", box_id=component.box_id, slot=component.slot_index),
}
def _build_duplicate_audit_payload(limit_groups: int = 60) -> dict:
"""构建重复物料巡检结果。
中文说明：本巡检不自动合并库存，只输出“疑似重复”清单和原因，
由人工在编辑页确认后执行合并，保持现有安全策略。
"""
boxes = Box.query.all()
box_by_id = {box.id: box for box in boxes}
enabled_components = Component.query.filter_by(is_enabled=True).all()
by_part_no = {}
by_material = {}
by_lcsc = {}
for component in enabled_components:
part_no_key = _normalize_part_no_key(component.part_no)
if part_no_key:
by_part_no.setdefault(part_no_key, []).append(component)
material_key = _material_identity_key(component.name, component.specification)
if material_key:
by_material.setdefault(material_key, []).append(component)
lcsc_code = _extract_lcsc_code_from_text(component.note or "")
if not lcsc_code:
lcsc_code = _extract_lcsc_code_from_text(component.part_no or "")
if lcsc_code:
by_lcsc.setdefault(lcsc_code, []).append(component)
def make_groups(source: dict, group_type: str, reason_text: str) -> list[dict]:
groups = []
for key, members in source.items():
if len(members) < 2:
continue
sorted_members = sorted(members, key=lambda c: (c.box_id, c.slot_index, c.id))
names = [(c.name or "").strip() for c in sorted_members]
specifications = [(c.specification or "").strip() for c in sorted_members]
unique_names = sorted({text for text in names if text})
unique_specifications = sorted({text for text in specifications if text})
standard_name = _pick_standard_text(names)
standard_specification = _pick_standard_text(specifications)
suggestion_lines = []
if len(unique_names) > 1 and standard_name:
suggestion_lines.append(f"建议统一名称为: {standard_name}")
if len(unique_specifications) > 1 and standard_specification:
suggestion_lines.append(f"建议统一规格为: {standard_specification}")
if not suggestion_lines:
suggestion_lines.append("命名和规格写法基本一致，可仅核对是否需要合并库存位置")
groups.append(
{
"type": group_type,
"key": key,
"reason": reason_text,
"member_count": len(sorted_members),
"unique_name_count": len(unique_names),
"unique_specification_count": len(unique_specifications),
"standard_name": standard_name,
"standard_specification": standard_specification,
"suggestion": "；".join(suggestion_lines),
"members": [_build_duplicate_member(component, box_by_id) for component in sorted_members],
}
)
groups.sort(key=lambda row: (-int(row["member_count"]), row["key"]))
return groups[:limit_groups]
part_no_groups = make_groups(by_part_no, "part_no", "疑似同料号")
material_groups = make_groups(by_material, "material", "疑似同参数")
lcsc_groups = make_groups(by_lcsc, "lcsc", "疑似同立创编号")
all_groups = part_no_groups + material_groups + lcsc_groups
all_groups.sort(key=lambda row: (-int(row["member_count"]), row["type"], row["key"]))
return {
"generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"total_enabled": len(enabled_components),
"total_groups": len(all_groups),
"part_no_groups": part_no_groups,
"material_groups": material_groups,
"lcsc_groups": lcsc_groups,
"groups": all_groups[:limit_groups],
}
def _build_duplicate_audit_summary_with_ai(payload: dict, settings: dict) -> tuple[str, str]:
api_key = (settings.get("api_key") or "").strip()
api_url = (settings.get("api_url") or "").strip()
model = (settings.get("model") or "").strip()
if not api_key or not api_url or not model:
return "", "AI 参数未完整配置，摘要使用规则生成"
brief_payload = {
"generated_at": payload.get("generated_at"),
"total_enabled": payload.get("total_enabled", 0),
"total_groups": payload.get("total_groups", 0),
"part_no_groups": len(payload.get("part_no_groups", [])),
"material_groups": len(payload.get("material_groups", [])),
"lcsc_groups": len(payload.get("lcsc_groups", [])),
"top_groups": [
{
"reason": g.get("reason", ""),
"key": g.get("key", ""),
"member_count": g.get("member_count", 0),
"suggestion": g.get("suggestion", ""),
}
for g in payload.get("groups", [])[:8]
],
}
system_prompt = (
"你是库存数据治理助手。"
"请输出简短中文总结（不超过120字），包含风险级别和处理优先顺序。"
"不要使用Markdown。"
)
user_prompt = "重复物料巡检结果(JSON):\n" + json.dumps(brief_payload, ensure_ascii=False)
try:
summary = _call_siliconflow_chat(
system_prompt,
user_prompt,
api_url=api_url,
model=model,
api_key=api_key,
timeout=int(settings.get("timeout", 30)),
)
return summary, ""
except Exception:
return "", "AI 摘要生成失败，已使用规则结果"
def _call_siliconflow_chat(
system_prompt: str,
user_prompt: str,
@@ -2849,6 +3027,154 @@ def ai_restock_plan():
}
@app.route("/ai/duplicate-audit", methods=["POST"])
def ai_duplicate_audit():
"""AI 重复物料巡检接口。
中文说明：
1. 巡检同料号、同参数、同立创编号三类疑似重复；
2. 输出重复原因与标准化建议；
3. 不自动合并，仅提供人工核对清单。
"""
payload = _build_duplicate_audit_payload(limit_groups=60)
settings = _get_ai_settings()
summary = ""
parse_warning = ""
if payload.get("total_groups", 0) > 0:
summary, parse_warning = _build_duplicate_audit_summary_with_ai(payload, settings)
if not summary:
part_no_count = len(payload.get("part_no_groups", []))
material_count = len(payload.get("material_groups", []))
lcsc_count = len(payload.get("lcsc_groups", []))
summary = (
"巡检完成: "
f"同料号 {part_no_count} 组，"
f"同参数 {material_count} 组，"
f"同立创编号 {lcsc_count} 组。"
"请优先处理成员数较多的分组。"
)
return {
"ok": True,
"summary": summary,
"parse_warning": parse_warning,
"data": payload,
}
@app.route("/ai/duplicate-audit/export")
def export_duplicate_audit_csv():
"""导出重复物料巡检结果 CSV。"""
limit_raw = (request.args.get("limit", "200") or "200").strip()
try:
limit_groups = int(limit_raw)
except ValueError:
limit_groups = 200
limit_groups = min(max(limit_groups, 1), 1000)
payload = _build_duplicate_audit_payload(limit_groups=limit_groups)
groups = payload.get("groups", [])
selected_group_ids = {
(group_id or "").strip()
for group_id in request.args.getlist("group_id")
if (group_id or "").strip()
}
if selected_group_ids:
groups = [
group
for group in groups
if f"{group.get('type', '')}::{group.get('key', '')}" in selected_group_ids
]
output = StringIO()
writer = csv.writer(output)
writer.writerow(
[
"巡检时间(generated_at)",
"重复类型(type)",
"重复原因(reason)",
"分组标识(key)",
"分组成员数(member_count)",
"建议(suggestion)",
"标准名称建议(standard_name)",
"标准规格建议(standard_specification)",
"元件ID(component_id)",
"料号(part_no)",
"名称(name)",
"规格(specification)",
"数量(quantity)",
"立创编号(lcsc_code)",
"盒子名称(box_name)",
"位置编号(slot_code)",
"编辑链接(edit_url)",
]
)
generated_at = payload.get("generated_at", "")
for group in groups:
members = group.get("members", []) or [None]
for member in members:
if member is None:
writer.writerow(
[
generated_at,
group.get("type", ""),
group.get("reason", ""),
group.get("key", ""),
group.get("member_count", 0),
group.get("suggestion", ""),
group.get("standard_name", ""),
group.get("standard_specification", ""),
"",
"",
"",
"",
"",
"",
"",
"",
"",
]
)
continue
writer.writerow(
[
generated_at,
group.get("type", ""),
group.get("reason", ""),
group.get("key", ""),
group.get("member_count", 0),
group.get("suggestion", ""),
group.get("standard_name", ""),
group.get("standard_specification", ""),
member.get("component_id", ""),
member.get("part_no", ""),
member.get("name", ""),
member.get("specification", ""),
member.get("quantity", 0),
member.get("lcsc_code", ""),
member.get("box_name", ""),
member.get("slot_code", ""),
member.get("edit_url", ""),
]
)
csv_content = "\ufeff" + output.getvalue()
output.close()
scope = "selected" if selected_group_ids else "all"
filename = f"duplicate_audit_{scope}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
return Response(
csv_content,
mimetype="text/csv; charset=utf-8",
headers={"Content-Disposition": f"attachment; filename={filename}"},
)
@app.route("/ai/settings", methods=["GET", "POST"])
def ai_settings_page():
settings = _get_ai_settings()

11
static/css/style.css
View File

@@ -370,6 +370,11 @@ body {
.side-low-stock-list li {
border: 1px solid var(--line);
border-radius: var(--radius);
.ai-divider {
border: 0;
border-top: 1px solid var(--line);
margin: var(--space-2) 0;
}
padding: 10px;
display: flex;
justify-content: space-between;
@@ -488,6 +493,12 @@ body {
font: 13px/1.55 Consolas, "Cascadia Mono", monospace;
}
.ai-divider {
border: 0;
border-top: 1px solid var(--line);
margin: var(--space-2) 0;
}
.ai-plan-groups {
display: grid;
gap: var(--space-2);

176
templates/types.html
View File

@@ -101,6 +101,21 @@
<summary>查看原始 AI 文本</summary>
<pre id="ai-panel-content" class="ai-panel-content"></pre>
</details>
<hr class="ai-divider">
<div class="ai-panel-head">
<h2>AI重复物料巡检</h2>
</div>
<p class="hint">扫描疑似同料号、同参数、同立创编号记录，生成人工复核清单。</p>
<div class="actions">
<button class="btn" id="ai-duplicate-btn" type="button">开始巡检</button>
<button class="btn btn-light" id="ai-duplicate-export-current-btn" type="button">导出当前显示</button>
<button class="btn btn-light" id="ai-duplicate-export-all-btn" type="button">导出全量</button>
</div>
<p class="hint" id="ai-duplicate-status"></p>
<p class="hint" id="ai-duplicate-warning"></p>
<div id="ai-duplicate-groups" class="ai-plan-groups"></div>
</section>
</aside>
</div>
@@ -244,6 +259,167 @@
});
});
})();
(function () {
var auditBtn = document.getElementById('ai-duplicate-btn');
var exportCurrentBtn = document.getElementById('ai-duplicate-export-current-btn');
var exportAllBtn = document.getElementById('ai-duplicate-export-all-btn');
var statusNode = document.getElementById('ai-duplicate-status');
var warningNode = document.getElementById('ai-duplicate-warning');
var groupsWrap = document.getElementById('ai-duplicate-groups');
var latestAuditGroups = [];
if (!auditBtn || !exportCurrentBtn || !exportAllBtn || !statusNode || !groupsWrap) {
return;
}
function clearGroups() {
latestAuditGroups = [];
groupsWrap.innerHTML = '';
}
function renderAuditGroups(groups) {
clearGroups();
if (!Array.isArray(groups) || !groups.length) {
var empty = document.createElement('p');
empty.className = 'muted';
empty.textContent = '未发现疑似重复物料';
groupsWrap.appendChild(empty);
return;
}
latestAuditGroups = groups.slice();
groups.forEach(function (group) {
var section = document.createElement('section');
section.className = 'ai-plan-group ai-audit-group';
var title = document.createElement('h3');
title.textContent = (group.reason || '疑似重复') + ' | ' + (group.key || '-') + '（' + (group.member_count || 0) + '）';
section.appendChild(title);
var suggestion = document.createElement('p');
suggestion.className = 'hint';
suggestion.textContent = '建议: ' + (group.suggestion || '请人工复核');
section.appendChild(suggestion);
var list = document.createElement('ul');
list.className = 'side-low-stock-list';
(group.members || []).forEach(function (member) {
var li = document.createElement('li');
var content = document.createElement('div');
var strong = document.createElement('strong');
strong.textContent = (member.name || '未命名元件') + ' (' + (member.part_no || '-') + ')';
content.appendChild(strong);
var spec = document.createElement('p');
spec.className = 'hint';
spec.textContent = '规格: ' + (member.specification || '-');
content.appendChild(spec);
var pos = document.createElement('p');
pos.className = 'hint';
pos.textContent = (member.box_name || '-') + ' / ' + (member.slot_code || '-') + ' | 数量 ' + (member.quantity || 0);
content.appendChild(pos);
if (member.lcsc_code) {
var lcsc = document.createElement('p');
lcsc.className = 'hint';
lcsc.textContent = '立创编号: ' + member.lcsc_code;
content.appendChild(lcsc);
}
li.appendChild(content);
if (member.edit_url) {
var editBtn = document.createElement('a');
editBtn.className = 'btn btn-light';
editBtn.href = member.edit_url;
editBtn.textContent = '编辑';
li.appendChild(editBtn);
}
list.appendChild(li);
});
section.appendChild(list);
groupsWrap.appendChild(section);
});
}
auditBtn.addEventListener('click', function () {
auditBtn.disabled = true;
statusNode.textContent = '正在巡检重复物料，请稍候...';
if (warningNode) {
warningNode.textContent = '';
}
clearGroups();
fetch('{{ url_for('ai_duplicate_audit') }}', {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: '{}'
})
.then(function (resp) {
return resp.json().then(function (data) {
return { ok: resp.ok, data: data };
});
})
.then(function (result) {
var data = result.data || {};
if (!result.ok || !data.ok) {
statusNode.textContent = '巡检失败';
if (warningNode) {
warningNode.textContent = data.message || '服务暂时不可用';
}
return;
}
statusNode.textContent = data.summary || '巡检已完成';
if (warningNode) {
warningNode.textContent = data.parse_warning || '';
}
renderAuditGroups((data.data && data.data.groups) || []);
})
.catch(function () {
statusNode.textContent = '巡检失败';
if (warningNode) {
warningNode.textContent = '请求失败，请稍后重试';
}
})
.finally(function () {
auditBtn.disabled = false;
});
});
exportCurrentBtn.addEventListener('click', function () {
if (!latestAuditGroups.length) {
statusNode.textContent = '请先运行巡检，再导出当前显示结果';
if (warningNode) {
warningNode.textContent = '';
}
return;
}
var params = new URLSearchParams();
params.set('limit', String(latestAuditGroups.length));
latestAuditGroups.forEach(function (group) {
params.append('group_id', (group.type || '') + '::' + (group.key || ''));
});
var downloadUrl = '{{ url_for('export_duplicate_audit_csv') }}?' + params.toString();
window.location.href = downloadUrl;
});
exportAllBtn.addEventListener('click', function () {
var downloadUrl = '{{ url_for('export_duplicate_audit_csv') }}?limit=1000';
window.location.href = downloadUrl;
});
})();
</script>
</body>
</html>