205 lines
14 KiB
Python
205 lines
14 KiB
Python
"""Collect official factual metadata. Run from repo root; no third-party packages.
|
||
HTML is cached outside the repository. No source prose or photos are persisted.
|
||
Mappings use FCI numbers and exact English names, never fuzzy breed matching.
|
||
"""
|
||
import argparse
|
||
from datetime import date
|
||
import concurrent.futures
|
||
import hashlib
|
||
import html
|
||
from html.parser import HTMLParser
|
||
import json
|
||
from pathlib import Path
|
||
import re
|
||
import unicodedata
|
||
import urllib.request
|
||
from urllib.parse import urljoin
|
||
|
||
ROOT = Path(__file__).resolve().parents[1] / 'internal/breeds'
|
||
CACHE = Path('/private/tmp/pawprint-breed-sources')
|
||
CACHE.mkdir(exist_ok=True)
|
||
DATE = date.today().isoformat()
|
||
|
||
class Node:
|
||
def __init__(self, tag='', attrs=None):
|
||
self.tag, self.attrs, self.children = tag, dict(attrs or []), []
|
||
def text(self):
|
||
return ''.join(c if isinstance(c, str) else c.text() for c in self.children)
|
||
def all(self):
|
||
yield self
|
||
for c in self.children:
|
||
if isinstance(c, Node):
|
||
yield from c.all()
|
||
|
||
class DOM(HTMLParser):
|
||
def __init__(self, text):
|
||
super().__init__(convert_charrefs=True)
|
||
self.root = Node(); self.stack = [self.root]; self.feed(text)
|
||
def handle_starttag(self, tag, attrs):
|
||
n = Node(tag, attrs); self.stack[-1].children.append(n)
|
||
if tag not in ('br','img','hr','meta','link','input','source','wbr','area','embed'):
|
||
self.stack.append(n)
|
||
def handle_startendtag(self, tag, attrs):
|
||
self.stack[-1].children.append(Node(tag,attrs))
|
||
def handle_endtag(self, tag):
|
||
for i in range(len(self.stack)-1,0,-1):
|
||
if self.stack[i].tag == tag:
|
||
self.stack = self.stack[:i]; return
|
||
def handle_data(self, data):
|
||
self.stack[-1].children.append(data)
|
||
|
||
def clean(value):
|
||
return re.sub(r'\s+', ' ', html.unescape(re.sub('<[^>]*>',' ',str(value or '')))).strip()
|
||
def norm(value):
|
||
return re.sub('[^a-z0-9]', '', unicodedata.normalize('NFKD', value).encode('ascii','ignore').decode().lower())
|
||
def fetch(url):
|
||
file = CACHE / (hashlib.sha256(url.encode()).hexdigest()+'.html')
|
||
if file.exists() and date.fromtimestamp(file.stat().st_mtime).isoformat()==DATE: return file.read_text()
|
||
req = urllib.request.Request(url, headers={'User-Agent':'PawprintPlan/1.0 (breed facts)'})
|
||
with urllib.request.urlopen(req, timeout=20) as r: text = r.read().decode('utf-8','replace')
|
||
file.write_text(text); return text
|
||
|
||
def references(provider, key, url, facts):
|
||
return {'provider':provider,'external_id':str(key),'url':url,'retrieved_at':DATE,
|
||
'facts':{k:clean(v) for k,v in facts.items() if v is not None and clean(v)}}
|
||
|
||
GROUPS=['牧羊犬与牧牛犬组','宾莎犬、雪纳瑞、獒犬与瑞士山地牧牛犬组','梗犬组','腊肠犬组','尖嘴犬与原始犬种组','嗅觉猎犬与相关犬种组','指示猎犬组','寻回犬、搜寻犬与水猎犬组','伴侣犬与玩具犬组','灵缇（视觉猎犬）组']
|
||
COUNTRIES={'AUSTRALIA':'澳大利亚','BELGIUM':'比利时','CROATIA':'克罗地亚','CZECHOSLOVAKIA':'捷克斯洛伐克','FRANCE':'法国','GERMANY':'德国','GREAT BRITAIN':'英国','HUNGARY':'匈牙利','ITALY':'意大利','POLAND':'波兰','PORTUGAL':'葡萄牙','ROMANIA':'罗马尼亚','RUSSIA':'俄罗斯','SLOVAKIA':'斯洛伐克','SPAIN':'西班牙','SWITZERLAND':'瑞士','THE NETHERLANDS':'荷兰','NETHERLANDS':'荷兰','AUSTRIA':'奥地利','DENMARK':'丹麦','FINLAND':'芬兰','IRELAND':'爱尔兰','NORWAY':'挪威','SWEDEN':'瑞典','USA':'美国','UNITED STATES OF AMERICA':'美国','CANADA':'加拿大','JAPAN':'日本','CHINA':'中国','KOREA':'韩国','ISRAEL':'以色列','MEXICO':'墨西哥','BRAZIL':'巴西','ARGENTINA':'阿根廷','AFGHANISTAN':'阿富汗','MIDDLE EAST':'中东','TURKEY':'土耳其','THAILAND':'泰国','SOUTH AFRICA':'南非','MOROCCO':'摩洛哥','MALI':'马里','ICELAND':'冰岛','GREENLAND':'格陵兰','MALTA':'马耳他','BOSNIA AND HERZEGOVINA':'波斯尼亚和黑塞哥维那','SERBIA':'塞尔维亚','MONTENEGRO':'黑山','GREECE':'希腊','URUGUAY':'乌拉圭','PERU':'秘鲁','INDONESIA':'印度尼西亚','LITHUANIA':'立陶宛','ESTONIA':'爱沙尼亚','BELGIUM, FRANCE':'比利时、法国','CENTRAL AFRICA':'非洲中部','CUBA':'古巴','CZECH REPUBLIC':'捷克','GERMANY, SWITZERLAND':'德国、瑞士','KOREA (REPUBLIC OF)':'韩国','NORTH MACEDONIA, SERBIA':'北马其顿、塞尔维亚','RUSSIAN FEDERATION':'俄罗斯','SLOVENIA':'斯洛文尼亚','TAIWAN':'中国台湾','TIBET (CHINA)':'中国西藏'}
|
||
|
||
def fci_group(link):
|
||
number, url=link
|
||
nodes=list(DOM(fetch(url)).root.all()); country=''; records={}
|
||
for n in nodes:
|
||
node_id=n.attrs.get('id','')
|
||
if re.search(r'_PaysLabel_\d+$',node_id): country=re.sub(r'^\d+\.\s*','',clean(n.text()))
|
||
if n.tag=='a' and 'RaceHyperLink' in node_id:
|
||
match=re.search(r'\((\d+)\)',n.text())
|
||
if not match: continue
|
||
key=match.group(1); title=clean(n.text()); en=re.findall(r'\(([^\d][^)]*)\)',title)
|
||
name = re.sub(r'-\d+\.html$', '', n.attrs['href'].split('/')[-1]).replace('-', ' ')
|
||
facts={'english_name':name.title(),'origin':COUNTRIES.get(country,country.title()),'breed_group':GROUPS[number-1],'fci_number':key}
|
||
records[key]=references('fci',key,urljoin(url,n.attrs['href']),facts)
|
||
return records
|
||
|
||
def akc_page(item):
|
||
name,url=item
|
||
nodes=list(DOM(fetch(url)).root.all()); props=None
|
||
for n in nodes:
|
||
if n.attrs.get('data-js-component')=='breedPage': props=json.loads(n.attrs['data-js-props']);break
|
||
if not props: return None
|
||
data=props['settings']['breed_data']; slug=props['settings']['current_breed']
|
||
basic=data.get('basics',{}).get(slug,{})
|
||
standard=data.get('standards',{}).get(slug,{})
|
||
traits=data.get('traits',{}).get(slug,{})
|
||
facts={'english_name':name,'life_span':basic.get('life_expectancy'),'height':standard.get('height_display'),'weight':standard.get('weight_display')}
|
||
scores={}
|
||
labels={'energy_level':'运动活力','trainability_level':'训练接受度','shedding_level':'掉毛程度','barking_level':'吠叫倾向','coat_grooming_frequency':'梳毛需求','affectionate_with_family':'亲近家人'}
|
||
for key, label in labels.items():
|
||
value=traits.get('traits',{}).get(key,{}).get('score')
|
||
if str(value) in ['1','2','3','4','5']:scores[label]=int(value)
|
||
if scores:facts['traits']=json.dumps(scores,ensure_ascii=False,separators=(',',':'))
|
||
words={'alert':'对环境敏锐','affectionate':'喜欢亲近家人','smart':'学习反应快','intelligent':'学习反应快','friendly':'亲人友善','active':'喜欢活动','energetic':'精力较充沛','playful':'喜欢玩耍','loyal':'重视与家人的联系','independent':'有自主性','gentle':'性情较温和','confident':'比较自信','courageous':'胆量较足','curious':'好奇心较强','calm':'较为沉着','outgoing':'乐于互动','devoted':'重视与家人的联系','lively':'喜欢活动','watchful':'对环境敏锐'}
|
||
tags=list(dict.fromkeys(words[w.strip().lower()] for w in traits.get('temperament','').split('/') if w.strip().lower() in words))
|
||
if tags:facts['temperament']='、'.join(tags)
|
||
return references('akc',slug,url,facts)
|
||
|
||
def localized(key, value):
|
||
value=clean(value).replace('(male)','（公犬）').replace('(female)','（母犬）').replace('up to','不超过').replace('years','年').replace('year','年')
|
||
unit, factor, label = ('inches',2.54,'cm') if key=='height' else ('pounds',0.45359237,'kg')
|
||
def convert(match):
|
||
low=f"{float(match.group(1))*factor:.1f}".rstrip('0').rstrip('.')
|
||
high=("–"+f"{float(match.group(2))*factor:.1f}".rstrip('0').rstrip('.')) if match.group(2) else ''
|
||
return low+high+' '+label
|
||
return re.sub(r'(\d+(?:\.\d+)?)(?:\s*[-–]\s*(\d+(?:\.\d+)?))?\s*'+unit,convert,value)
|
||
|
||
def main():
|
||
parser=argparse.ArgumentParser();parser.parse_args()
|
||
rows=json.loads((ROOT/'data.json').read_text())
|
||
home=list(DOM(fetch('https://www.fci.be/en/nomenclature/')).root.all())
|
||
links=[]
|
||
for n in home:
|
||
if n.tag=='a' and 'GroupesRepeater_GroupeHyperLink' in n.attrs.get('id',''):
|
||
links.append((int(clean(n.text()).split()[-1]),urljoin('https://www.fci.be',n.attrs['href'])))
|
||
fci={}
|
||
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool:
|
||
for result in pool.map(fci_group,links): fci.update(result)
|
||
print('FCI catalog:',len(fci),flush=True)
|
||
akc={}
|
||
for n in DOM(fetch('https://www.akc.org/dog-breeds/')).root.all():
|
||
if n.tag=='option' and re.fullmatch(r'https://www.akc.org/dog-breeds/[a-z0-9-]+/',n.attrs.get('value','')):
|
||
akc[norm(clean(n.text()))]=(clean(n.text()),n.attrs['value'])
|
||
matches={}; entries=[]
|
||
for row in rows:
|
||
refs=[];fields={};field_sources={}
|
||
existing={k:row.get(k,'') for k in ['english_name','origin','breed_group','fci_number','height','weight','temperament','traits'] if row.get(k)}
|
||
if row['source_key'].startswith('cku:'):
|
||
refs.append(references('cku',row['source_key'].split(':')[1],row['source_url'],{'name':row['name'],'aliases':row['aliases'],**existing}))
|
||
for k in existing:field_sources[k]='cku'
|
||
match=re.match(r'cku:(\d+)',row['source_key']); source=fci.get(match.group(1)) if match else None
|
||
if source:
|
||
source=json.loads(json.dumps(source))
|
||
if re.search(r'[A-Z]$',row['source_key']):source['facts']['scope']='犬种基础标准（包含多个变种）'
|
||
refs.append(source)
|
||
for k,v in source['facts'].items():
|
||
if k not in ['scope'] and not (k=='english_name' and re.search(r'[A-Z]$',row['source_key'])) and not row.get(k): fields[k]=v;field_sources[k]='fci'
|
||
candidates=[row.get('english_name',''),fields.get('english_name','')]
|
||
equivalents={'cku:39':'Pembroke Welsh Corgi','cku:38':'Cardigan Welsh Corgi','cku:167':'Cocker Spaniel'}
|
||
if row['source_key'] in equivalents:candidates.insert(0,equivalents[row['source_key']])
|
||
# Varieties are deliberately not collapsed into their parent breed.
|
||
for candidate in candidates:
|
||
if norm(candidate) in akc:
|
||
item=akc[norm(candidate)];matches[item]=None;break
|
||
entries.append({'source_key':row['source_key'],'name':row['name'],'fields':fields,'field_sources':field_sources,'references':refs})
|
||
print('Exact AKC matches:',len(matches),flush=True)
|
||
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
|
||
futures={pool.submit(akc_page,item):item for item in matches}
|
||
done=0
|
||
for job in concurrent.futures.as_completed(futures):
|
||
item=futures[job]
|
||
try:matches[item]=job.result()
|
||
except Exception as e: print('AKC unavailable:',item[0],type(e).__name__,flush=True)
|
||
done+=1
|
||
if done%20==0:print('AKC progress:',done,'/',len(matches),flush=True)
|
||
by_name={norm(name):ref for (name,_),ref in matches.items() if ref}
|
||
for row,entry in zip(rows,entries):
|
||
names=[row.get('english_name',''),entry['fields'].get('english_name','')]
|
||
if row['source_key'] in equivalents:names.insert(0,equivalents[row['source_key']])
|
||
ref=next((by_name[norm(n)] for n in names if norm(n) in by_name),None)
|
||
if not ref:continue
|
||
ref=json.loads(json.dumps(ref))
|
||
if re.search(r'[A-Z]$',row['source_key']):
|
||
ref['facts']['scope']='父犬种资料，变种差异以对应标准为准'
|
||
entry['references'].append(ref);continue
|
||
entry['references'].append(ref)
|
||
for k,v in ref['facts'].items():
|
||
if k in ['height','weight','life_span'] or (not row.get(k) and not entry['fields'].get(k)):entry['fields'][k]=localized(k,v);entry['field_sources'][k]='akc'
|
||
# Lifestyle ratings have an explicit AKC source, keeping CKU ratings in the reference.
|
||
if ref['facts'].get('traits'):entry['fields']['traits']=ref['facts']['traits'];entry['field_sources']['traits']='akc'
|
||
rkc={r['source_key']:r for r in json.loads((ROOT/'rkc_facts.json').read_text())}
|
||
for row,entry in zip(rows,entries):
|
||
if row['source_key'] in rkc:
|
||
r=rkc[row['source_key']]
|
||
rkc_ref=references('rkc',r['external_id'],r['url'],r['facts']);rkc_ref['retrieved_at']=r['retrieved_at'];entry['references'].append(rkc_ref)
|
||
for k in ['exercise','grooming','life_span']:
|
||
if k=='life_span' and entry['fields'].get(k):continue
|
||
entry['fields'][k]=r['facts'][k];entry['field_sources'][k]='rkc'
|
||
if not entry['fields'].get('summary') and not row.get('summary'):
|
||
facts={**row,**entry['fields']};bits=[]
|
||
if facts.get('origin'):bits.append('这一犬种起源于'+facts['origin'])
|
||
if facts.get('breed_group'):bits.append('在 FCI 体系中归入'+facts['breed_group'])
|
||
if facts.get('temperament'):bits.append('常见性格倾向包括'+facts['temperament'])
|
||
if bits:entry['fields']['summary']='。'.join(bits)+'。';entry['field_sources']['summary']='editorial'
|
||
if row.get('summary'):entry['field_sources']['summary']='editorial'
|
||
if row.get('care_notes'):entry['field_sources']['care_notes']='editorial'
|
||
if entry['references']:
|
||
scores={}
|
||
try:scores=json.loads(entry['fields'].get('traits') or row.get('traits') or '{}')
|
||
except ValueError:pass
|
||
if not entry['fields'].get('exercise') and scores.get('运动活力'):
|
||
entry['fields']['exercise']='运动活力指数 '+str(scores['运动活力'])+'/5，结合年龄与体力安排散步和互动';entry['field_sources']['exercise']='editorial'
|
||
if not entry['fields'].get('grooming') and scores.get('梳毛需求'):
|
||
entry['fields']['grooming']='梳毛需求指数 '+str(scores['梳毛需求'])+'/5，按被毛情况安排梳理';entry['field_sources']['grooming']='editorial'
|
||
(ROOT/'enrichment.json').write_text(json.dumps(entries,ensure_ascii=False,indent=2)+'\n')
|
||
print('Saved',len(entries),'breeds,',sum(len(e['references']) for e in entries),'references',flush=True)
|
||
if __name__=='__main__':main()
|