|
|
@@ -1,355 +0,0 @@
|
|
|
-import sys, os, csv, re
|
|
|
-sys.stdout.reconfigure(encoding='utf-8')
|
|
|
-sys.stderr.reconfigure(encoding='utf-8')
|
|
|
-from PyPDF2 import PdfReader
|
|
|
-
|
|
|
-path = r'D:\workspace\cfc\docs\参考资料'
|
|
|
-outdir = path
|
|
|
-columns = ['名称', '分类', '推荐指数', '能量KJ', '蛋白g', '脂肪g', '碳水化合物g', '淀粉g', '总膳食纤维g', '胆固醇mg']
|
|
|
-known_cats = ['主食', '乳制品', '干果', '坚果', '快餐', '水产品', '水果', '汤', '肉类', '蔬菜', '豆类及豆制品', '蛋类', '饮料']
|
|
|
-
|
|
|
-# CJK Radical -> Standard Hanzi mapping (observed in these PDFs)
|
|
|
-_RADICAL_MAP = {
|
|
|
- '\u2f24': '大', # ⼤ -> 大
|
|
|
- '\u2f29': '小', # ⼩ -> 小
|
|
|
- '\u2f2d': '山', # ⼭ -> 山
|
|
|
- '\u2f32': '干', # ⼲ -> 干
|
|
|
- '\u2f46': '无', # ⽆ -> 无
|
|
|
- '\u2f4a': '木', # ⽊ -> 木
|
|
|
- '\u2f50': '比', # ⽐ -> 比
|
|
|
- '\u2f54': '水', # ⽔ -> 水
|
|
|
- '\u2f55': '火', # ⽕ -> 火
|
|
|
- '\u2f5c': '牛', # ⽜ -> 牛
|
|
|
- '\u2f5f': '玉', # ⽟ -> 玉
|
|
|
- '\u2f62': '甘', # ⽢ -> 甘
|
|
|
- '\u2f63': '生', # ⽣ -> 生
|
|
|
- '\u2f64': '用', # ⽤ -> 用
|
|
|
- '\u2f69': '白', # ⽩ -> 白
|
|
|
- '\u2f6f': '石', # ⽯ -> 石
|
|
|
- '\u2f75': '竹', # ⽵ -> 竹
|
|
|
- '\u2f76': '米', # ⽶ -> 米
|
|
|
- '\u2f7b': '羽', # ⽻ -> 羽
|
|
|
- '\u2f7c': '老', # ⽼ -> 老
|
|
|
- '\u2f81': '肉', # ⾁ -> 肉
|
|
|
- '\u2f96': '豆', # ⾖ -> 豆
|
|
|
- '\u2f9d': '身', # ⾝ -> 身
|
|
|
- '\u2faf': '面', # ⾯ -> 面
|
|
|
- '\u2fb2': '韭', # ⾲ -> 韭
|
|
|
- '\u2fb9': '香', # ⾹ -> 香
|
|
|
- '\u2fca': '黑', # ⿊ -> 黑
|
|
|
- '\u2ee5': '鱼', # ⻥ -> 鱼
|
|
|
- '\u2ee8': '麦', # ⻨ -> 麦
|
|
|
- '\u2ee9': '黄', # ⻩ -> 黄
|
|
|
- '\u2ec9': '贝', # ⻉ -> 贝
|
|
|
- '\u2ee2': '马', # ⻢ -> 马
|
|
|
- '\u2edd': '食', # ⻝ -> 食
|
|
|
- '\u2ef0': '龙', # ⻰ -> 龙
|
|
|
- '\u2f18': '卜', # ⼘ -> 卜
|
|
|
- '\u2f1f': '土', # ⼟ -> 土
|
|
|
- '\u2f26': '子', # ⼦ -> 子
|
|
|
- '\u2f3c': '心', # ⼼ -> 心
|
|
|
- '\u2f42': '文', # ⽂ -> 文
|
|
|
- '\u2f60': '瓜', # ⽠ -> 瓜
|
|
|
- '\u2f6a': '皮', # ⽪ -> 皮
|
|
|
- '\u2f6c': '目', # ⽬ -> 目
|
|
|
- '\u2f7a': '羊', # ⽺ -> 羊
|
|
|
- '\u2f7f': '耳', # ⽿ -> 耳
|
|
|
- '\u2f90': '衣', # ⾐ -> 衣
|
|
|
- '\u2f95': '谷', # ⾕ -> 谷
|
|
|
- '\u2fa6': '金', # ⾦ -> 金
|
|
|
-}
|
|
|
-
|
|
|
-def normalize_text(text):
|
|
|
- """Replace CJK radical characters with standard Hanzi."""
|
|
|
- result = []
|
|
|
- for ch in text:
|
|
|
- result.append(_RADICAL_MAP.get(ch, ch))
|
|
|
- return ''.join(result)
|
|
|
-
|
|
|
-# Load reference (侯 report) for nutritional values
|
|
|
-ref_path = os.path.join(path, '530010234-侯-个体化食物推荐表.csv')
|
|
|
-ref_nutrition = {} # name -> {field: value}
|
|
|
-with open(ref_path, 'r', encoding='utf-8-sig') as f:
|
|
|
- for row in csv.DictReader(f):
|
|
|
- ref_nutrition[row['名称']] = {k: row[k] for k in ['能量KJ','蛋白g','脂肪g','碳水化合物g','淀粉g','总膳食纤维g','胆固醇mg']}
|
|
|
-
|
|
|
-def parse_compressed_v2(text):
|
|
|
- """Parse compressed format (朱评估报告 style) using reference nutrition values."""
|
|
|
- # Remove header
|
|
|
- header = '名称分类推荐指数能量KJ蛋白g脂肪g碳水化合物g淀粉g总膳食纤维g胆固醇mg'
|
|
|
- rest = text
|
|
|
- if header in rest:
|
|
|
- rest = rest.replace(header, '')
|
|
|
-
|
|
|
- # Remove known junk patterns
|
|
|
- skip_kws = ['根据您的肠道菌群', '分值从-100', '食物推荐考虑', '食物推荐是综合',
|
|
|
- '需要注意的是', '本饮食推荐', '该饮食推荐根据', '后续表格中的营养',
|
|
|
- '肠道菌群健康检测报告说明', '检测方法及局限性', '数据分析及模型',
|
|
|
- '本报告中的检测', '以上模型预测', '结果解读及使用', '正常范围的定义',
|
|
|
- '影响因素说明', '建议将检测结果', '营养建议说明', '报告中的食物推荐',
|
|
|
- '重要提示', '推荐食物清单', '实际食用时需结合', '如有特殊疾病',
|
|
|
- '免责声明', '本检测报告仅供', '16S 高通量测序', '基于机器学习和']
|
|
|
- for kw in skip_kws:
|
|
|
- rest = rest.replace(kw, '')
|
|
|
-
|
|
|
- rows = []
|
|
|
-
|
|
|
- while rest:
|
|
|
- # Find earliest category
|
|
|
- best_cat, best_idx = None, len(rest)
|
|
|
- for cat in known_cats:
|
|
|
- idx = rest.find(cat)
|
|
|
- if idx != -1 and idx < best_idx:
|
|
|
- best_idx, best_cat = idx, cat
|
|
|
- if best_cat is None:
|
|
|
- break
|
|
|
-
|
|
|
- name = rest[:best_idx]
|
|
|
- rest = rest[best_idx + len(best_cat):]
|
|
|
-
|
|
|
- # Extract digit string
|
|
|
- num_str = ''
|
|
|
- while rest and (rest[0].isdigit() or rest[0] in '-\u2212\u2014'):
|
|
|
- c = rest[0]
|
|
|
- if c in '\u2212\u2014':
|
|
|
- c = '-'
|
|
|
- num_str += c
|
|
|
- rest = rest[1:]
|
|
|
-
|
|
|
- if not name or not num_str:
|
|
|
- continue
|
|
|
-
|
|
|
- # Try to decode using reference
|
|
|
- decoded = decode_num_str(name, num_str)
|
|
|
- if decoded:
|
|
|
- rows.append(dict(zip(columns, [name, best_cat] + [str(v) for v in decoded])))
|
|
|
-
|
|
|
- return rows
|
|
|
-
|
|
|
-def decode_num_str(name, s):
|
|
|
- """Decode 8-field number string using reference nutrition values."""
|
|
|
- # Expected 7 nutrition values from reference
|
|
|
- ref_vals = ref_nutrition.get(name)
|
|
|
- if ref_vals:
|
|
|
- ref_nums = [int(ref_vals[k]) for k in ['能量KJ','蛋白g','脂肪g','碳水化合物g','淀粉g','总膳食纤维g','胆固醇mg']]
|
|
|
- else:
|
|
|
- ref_nums = None
|
|
|
-
|
|
|
- raw = s.lstrip('-') # remove leading dash from 推荐指数
|
|
|
- has_neg = s.startswith('-')
|
|
|
-
|
|
|
- results = []
|
|
|
-
|
|
|
- # Try all possible splits for 推荐指数 (1 or 2 digits) and 能量KJ (2-4 digits)
|
|
|
- for rec_len in range(1, 3):
|
|
|
- if rec_len > len(raw):
|
|
|
- continue
|
|
|
- rec_str = '-' + raw[:rec_len] if has_neg else raw[:rec_len]
|
|
|
- rec_val = int(rec_str)
|
|
|
- if not (-100 <= rec_val <= 100):
|
|
|
- continue
|
|
|
-
|
|
|
- # Now try to split the remaining 7 fields
|
|
|
- remain = raw[rec_len:]
|
|
|
- candidates = split_7_fields(remain)
|
|
|
-
|
|
|
- for nums in candidates:
|
|
|
- full = [rec_val] + nums
|
|
|
- if len(full) == 8:
|
|
|
- # If we have reference values, check match (allow 1-char shifts)
|
|
|
- if ref_nums:
|
|
|
- matches = sum(1 for i in range(7) if ref_nums[i] == nums[i])
|
|
|
- if matches >= 6:
|
|
|
- results.append(full)
|
|
|
- else:
|
|
|
- results.append(full)
|
|
|
-
|
|
|
- if not results:
|
|
|
- return None
|
|
|
-
|
|
|
- # Pick best: max nutrition matches, then shortest rec_len
|
|
|
- if ref_nums:
|
|
|
- def score(r):
|
|
|
- matches = sum(1 for i in range(7) if ref_nums[i] == r[1:][i])
|
|
|
- # Penalize rec_len=2 slightly (prefer 1-digit rec)
|
|
|
- return (matches, -len(str(abs(r[0]))))
|
|
|
- results.sort(key=score, reverse=True)
|
|
|
-
|
|
|
- r = results[0]
|
|
|
- if ref_nums:
|
|
|
- matches = sum(1 for i in range(7) if ref_nums[i] == r[1:][i])
|
|
|
- if matches < 6:
|
|
|
- return None
|
|
|
-
|
|
|
- return r
|
|
|
-
|
|
|
-def split_7_fields(s):
|
|
|
- """Split remaining string into 7 numeric fields using width heuristics."""
|
|
|
- results = []
|
|
|
- # Expected widths (min, max) for each field after 能量KJ
|
|
|
- # 能量KJ: 2-4, 蛋白: 1-2, 脂肪: 1-2, 碳水: 1-2, 淀粉: 1-2, 膳食纤维: 1-2, 胆固醇: 1-4
|
|
|
- ranges = [(2,4), (1,2), (1,2), (1,2), (1,2), (1,2), (1,4)]
|
|
|
-
|
|
|
- def backtrack(pos, field_idx, nums):
|
|
|
- if field_idx == 7:
|
|
|
- if pos == len(s):
|
|
|
- results.append(list(nums))
|
|
|
- return
|
|
|
- if pos >= len(s):
|
|
|
- return
|
|
|
- min_w, max_w = ranges[field_idx]
|
|
|
- for w in range(min_w, min(max_w + 1, len(s) - pos + 1)):
|
|
|
- chunk = s[pos:pos+w]
|
|
|
- if chunk.isdigit():
|
|
|
- val = int(chunk)
|
|
|
- # 胆固醇 can be up to 4 digits (>999), allow common values
|
|
|
- if field_idx == 0: # 能量KJ - typically >= 100 for meaningful foods
|
|
|
- pass # allow all
|
|
|
- backtrack(pos+w, field_idx+1, nums+[val])
|
|
|
-
|
|
|
- backtrack(0, 0, [])
|
|
|
- return results
|
|
|
-
|
|
|
-def parse_vertical(all_lines):
|
|
|
- rows = []
|
|
|
- i = 0
|
|
|
- while i + 9 < len(all_lines):
|
|
|
- name = all_lines[i].strip()
|
|
|
- cat = all_lines[i+1].strip()
|
|
|
- if cat not in known_cats:
|
|
|
- i += 1
|
|
|
- continue
|
|
|
- nums = []
|
|
|
- ok = True
|
|
|
- for j in range(2, 10):
|
|
|
- try:
|
|
|
- v = all_lines[i+j].replace('\u2212', '-').replace('\u2014', '-').strip()
|
|
|
- int(v)
|
|
|
- nums.append(v)
|
|
|
- except:
|
|
|
- ok = False
|
|
|
- break
|
|
|
- if ok and len(nums) == 8:
|
|
|
- rows.append(dict(zip(columns, [name, cat] + nums)))
|
|
|
- i += 1
|
|
|
- return rows
|
|
|
-
|
|
|
-def detect_format(text):
|
|
|
- lines = [l.strip() for l in text.split('\n') if l.strip() and not re.match(r'^\d+/\d+$', l)]
|
|
|
- if not lines:
|
|
|
- return 'unknown'
|
|
|
- long_line_count = sum(1 for l in lines[:10] if len(l) > 100)
|
|
|
- cat_count = sum(1 for l in lines[:80] if l.strip() in known_cats)
|
|
|
-
|
|
|
- if len(lines) <= 3 and len(lines[0]) > 200:
|
|
|
- return 'compressed'
|
|
|
- if long_line_count >= 3:
|
|
|
- return 'compressed'
|
|
|
- if cat_count >= 5:
|
|
|
- return 'vertical'
|
|
|
- return 'unknown'
|
|
|
-
|
|
|
-for pdf_file in sorted(os.listdir(path)):
|
|
|
- if not pdf_file.endswith('.pdf'):
|
|
|
- continue
|
|
|
- report_name = pdf_file.replace('.pdf', '')
|
|
|
- pdf_path = os.path.join(path, pdf_file)
|
|
|
- try:
|
|
|
- reader = PdfReader(pdf_path)
|
|
|
- except:
|
|
|
- print(f'ERROR {report_name}')
|
|
|
- continue
|
|
|
-
|
|
|
- food_start = None
|
|
|
- for i, page in enumerate(reader.pages):
|
|
|
- if '个体化食物推荐表' in page.extract_text():
|
|
|
- food_start = i
|
|
|
- break
|
|
|
- if food_start is None:
|
|
|
- print(f'SKIP {report_name}')
|
|
|
- continue
|
|
|
-
|
|
|
- print(f'{report_name}...', end=' ')
|
|
|
- first_data = normalize_text(reader.pages[food_start + 1].extract_text())
|
|
|
- fmt = detect_format(first_data)
|
|
|
- print(f'[{fmt}]', end=' ')
|
|
|
-
|
|
|
- food_rows = []
|
|
|
- if fmt == 'compressed':
|
|
|
- for i in range(food_start + 1, len(reader.pages)):
|
|
|
- text = normalize_text(reader.pages[i].extract_text())
|
|
|
- # Remove page numbers
|
|
|
- text = re.sub(r'\d+/\d+', '', text)
|
|
|
- food_rows.extend(parse_compressed_v2(text))
|
|
|
- else:
|
|
|
- all_lines = []
|
|
|
- for i in range(food_start + 1, len(reader.pages)):
|
|
|
- for line in normalize_text(reader.pages[i].extract_text()).split('\n'):
|
|
|
- lt = line.strip()
|
|
|
- if not lt or re.match(r'^\d+/\d+$', lt) or lt in columns:
|
|
|
- continue
|
|
|
- if len(lt) > 60 and any(k in lt for k in ['根据您的', '分值从', '食物推荐', '需要注意', '本饮食推荐', '该饮食推荐', '后续表格']):
|
|
|
- continue
|
|
|
- all_lines.append(lt)
|
|
|
- food_rows = parse_vertical(all_lines)
|
|
|
-
|
|
|
- print(f'{len(food_rows)} entries')
|
|
|
-
|
|
|
- if food_rows:
|
|
|
- csv_name = f'{report_name}-个体化食物推荐表.csv'
|
|
|
- csv_path = os.path.join(outdir, csv_name)
|
|
|
- with open(csv_path, 'w', newline='', encoding='utf-8-sig') as f:
|
|
|
- writer = csv.DictWriter(f, fieldnames=columns)
|
|
|
- writer.writeheader()
|
|
|
- writer.writerows(food_rows)
|
|
|
- print(f' -> {csv_name}')
|
|
|
-
|
|
|
-# ── Step 2: 生成推荐指数汇总表(标准营养 + 每人推荐指数) ──
|
|
|
-print('\n--- 生成推荐指数汇总表 ---')
|
|
|
-
|
|
|
-# 用侯报告作为标准营养参考(224条最完整)
|
|
|
-ref_csv = os.path.join(outdir, '530010234-侯-个体化食物推荐表.csv')
|
|
|
-if os.path.exists(ref_csv):
|
|
|
- with open(ref_csv, 'r', encoding='utf-8-sig') as f:
|
|
|
- std_rows = list(csv.DictReader(f))
|
|
|
- std_lookup = {}
|
|
|
- for r in std_rows:
|
|
|
- name = r['名称']
|
|
|
- std_lookup[name] = {k: r[k] for k in ['分类','能量KJ','蛋白g','脂肪g','碳水化合物g','淀粉g','总膳食纤维g','胆固醇mg']}
|
|
|
-
|
|
|
- # 收集每份报告的推荐指数
|
|
|
- rec_lookups = {}
|
|
|
- report_labels = {'501999942-某人-个体化食物推荐表.csv': '某人',
|
|
|
- '530010234-侯-个体化食物推荐表.csv': '侯',
|
|
|
- '547982403-个体化食物推荐表.csv': '547982403',
|
|
|
- '儿童示例-个体化食物推荐表.csv': '儿童示例',
|
|
|
- '朱评估报告-个体化食物推荐表.csv': '朱评估报告'}
|
|
|
- for csv_name, label in report_labels.items():
|
|
|
- p = os.path.join(outdir, csv_name)
|
|
|
- if os.path.exists(p):
|
|
|
- with open(p, 'r', encoding='utf-8-sig') as f:
|
|
|
- for row in csv.DictReader(f):
|
|
|
- rec_lookups.setdefault(label, {})[row['名称']] = row['推荐指数']
|
|
|
-
|
|
|
- sum_cols = ['名称','分类','能量KJ','蛋白g','脂肪g','碳水化合物g','淀粉g','总膳食纤维g','胆固醇mg',
|
|
|
- '某人','侯','547982403','儿童示例','朱评估报告']
|
|
|
- sum_path = os.path.join(outdir, '推荐指数汇总.csv')
|
|
|
- with open(sum_path, 'w', newline='', encoding='utf-8-sig') as f:
|
|
|
- w = csv.DictWriter(f, fieldnames=sum_cols)
|
|
|
- w.writeheader()
|
|
|
- for r in std_rows:
|
|
|
- name = r['名称']
|
|
|
- s = std_lookup[name]
|
|
|
- row = {'名称': name, '分类': s['分类'],
|
|
|
- '能量KJ': s['能量KJ'], '蛋白g': s['蛋白g'], '脂肪g': s['脂肪g'],
|
|
|
- '碳水化合物g': s['碳水化合物g'], '淀粉g': s['淀粉g'],
|
|
|
- '总膳食纤维g': s['总膳食纤维g'], '胆固醇mg': s['胆固醇mg']}
|
|
|
- for label in ['某人','侯','547982403','儿童示例','朱评估报告']:
|
|
|
- row[label] = rec_lookups.get(label, {}).get(name, '')
|
|
|
- w.writerow(row)
|
|
|
- print(f'推荐指数汇总表: {len(std_rows)} 行 -> {sum_path}')
|
|
|
- for label in ['某人','侯','547982403','儿童示例','朱评估报告']:
|
|
|
- missing = sum(1 for r in std_rows if not rec_lookups.get(label, {}).get(r['名称'], ''))
|
|
|
- print(f' {label}: 缺失 {missing} 条')
|
|
|
-
|
|
|
-print('\nDone!')
|