92 行
3.9 KiB
Python
92 行
3.9 KiB
Python
"""Download a catalog-linked datasheet, retaining its source URL and SHA256."""
|
|
import argparse
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
from html.parser import HTMLParser
|
|
import json
|
|
from pathlib import Path
|
|
from urllib.parse import urljoin, urlsplit
|
|
|
|
import fitz
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
class Links(HTMLParser):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.pdfs = []
|
|
|
|
def handle_starttag(self, tag, attrs):
|
|
for key, value in attrs:
|
|
if key in {'src', 'href', 'data'} and value and urlsplit(value).path.lower().endswith('.pdf'):
|
|
self.pdfs.append(value.split('#')[0])
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(description=__doc__)
|
|
ap.add_argument('name')
|
|
ap.add_argument('url')
|
|
ap.add_argument('--html', action='store_true')
|
|
ap.add_argument('--out-dir', type=Path, default=ROOT/'datasheets/fans')
|
|
args = ap.parse_args()
|
|
if not all(c.isalnum() or c in '-_.' for c in args.name):
|
|
raise ValueError('Filename must be a simple part name')
|
|
output = args.out_dir.resolve()
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
session = requests.Session()
|
|
session.trust_env = False
|
|
response = session.get(args.url, timeout=25)
|
|
response.raise_for_status()
|
|
page = response
|
|
if args.html:
|
|
soup = BeautifulSoup(response.text, 'html.parser')
|
|
path = output/(args.name+'.html')
|
|
path.write_bytes(response.content)
|
|
text = soup.get_text('\n', strip=True)
|
|
path.with_suffix('.txt').write_text(text, encoding='utf-8')
|
|
images = []
|
|
for tag in soup.find_all('img', src=True):
|
|
if any(key in tag['src'] for key in ['pwm_02_en', 'sensor_01_en', 'sensor_02_en']):
|
|
image_url = urljoin(response.url, tag['src'])
|
|
image = session.get(image_url, timeout=20)
|
|
image.raise_for_status()
|
|
image_path = output/Path(urlsplit(image_url).path).name
|
|
image_path.write_bytes(image.content)
|
|
images.append({'url': image_url, 'file': image_path.name,
|
|
'sha256': hashlib.sha256(image.content).hexdigest()})
|
|
metadata = {'retrieved_at_utc': datetime.now(timezone.utc).isoformat(),
|
|
'requested_url': args.url, 'source_page_url': response.url,
|
|
'sha256': hashlib.sha256(response.content).hexdigest(), 'images': images}
|
|
path.with_suffix('.json').write_text(json.dumps(metadata, indent=2)+'\n', encoding='utf-8')
|
|
print(json.dumps(metadata, indent=2))
|
|
return
|
|
if not response.content.startswith(b'%PDF'):
|
|
parser = Links()
|
|
parser.feed(response.text)
|
|
urls = list(dict.fromkeys(urljoin(response.url, p) for p in parser.pdfs))
|
|
if len(urls) != 1:
|
|
raise ValueError(f'Expected a unique linked PDF; found {urls}')
|
|
(output/(args.name+'-source.html')).write_text(response.text, encoding='utf-8')
|
|
response = session.get(urls[0], timeout=25)
|
|
response.raise_for_status()
|
|
if not response.content.startswith(b'%PDF'):
|
|
raise ValueError('Response is not a PDF')
|
|
document = fitz.open(stream=response.content, filetype='pdf')
|
|
pdf = output/(args.name+'.pdf')
|
|
pdf.write_bytes(response.content)
|
|
text = '\n'.join(f'\n--- PDF PAGE {p.number+1} ---\n'+p.get_text() for p in document)
|
|
pdf.with_suffix('.txt').write_text(text, encoding='utf-8')
|
|
metadata = {'retrieved_at_utc': datetime.now(timezone.utc).isoformat(), 'requested_url': args.url,
|
|
'source_page_url': page.url, 'pdf_url': response.url, 'pages': len(document),
|
|
'sha256': hashlib.sha256(response.content).hexdigest(), 'bytes': len(response.content)}
|
|
pdf.with_suffix('.json').write_text(json.dumps(metadata, indent=2)+'\n', encoding='utf-8')
|
|
print(json.dumps(metadata, indent=2))
|
|
print(text[:1600])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|