MVP сайта

examples/parsers/common.py

@@ -0,0 +1,16 @@
+import json
+from pathlib import Path
+from bs4 import BeautifulSoup
+
+def extract(markup):
+    soup=BeautifulSoup(markup,'html.parser')
+    records=[]
+    def field_data(field):
+        return {'pointer':field['data-field'],'label':field.get('data-label',''),'state':field['data-value-state'],'type':field.get('data-value-type','string'),'text':field.get_text()}
+    for record in soup.select('.record[data-record-id]'):
+        records.append({'id':record['data-record-id'],'fields':[field_data(f) for f in record.select('.field[data-field]:not([data-role])')],
+                        'roles':{f['data-role']:field_data(f) for f in record.select('.field[data-role]')}})
+    return records
+
+def save(records,path):
+    Path(path).write_text(''.join(json.dumps(r,ensure_ascii=False)+'\n' for r in records),encoding='utf-8')

examples/parsers/scrape_html.py

@@ -0,0 +1,26 @@
+"""python scrape_html.py URL --output records.jsonl"""
+import argparse
+from urllib.parse import urljoin,urlsplit
+import requests
+from bs4 import BeautifulSoup
+from common import extract,save
+
+def scrape(url):
+    records=[]; pages=set(); ids=set(); origin=urlsplit(url).netloc
+    with requests.Session() as session:
+        while url:
+            if url in pages: raise RuntimeError('Зацикленная пагинация')
+            pages.add(url)
+            response=session.get(url,timeout=30); response.raise_for_status()
+            batch=extract(response.text)
+            for record in batch:
+                if record['id'] in ids: raise RuntimeError('Повтор записи между страницами')
+                ids.add(record['id']); records.append(record)
+            next_link=BeautifulSoup(response.text,'html.parser').select_one('a[rel=next]')
+            url=urljoin(url,next_link['href']) if next_link else ''
+            if url and urlsplit(url).netloc!=origin: raise RuntimeError('Ссылка ведёт на другой сервер')
+    return records
+
+if __name__=='__main__':
+    p=argparse.ArgumentParser();p.add_argument('url');p.add_argument('--output',default='records.jsonl');a=p.parse_args()
+    records=scrape(a.url);save(records,a.output);print(f'Сохранено записей: {len(records)}')

examples/parsers/scrape_interactive.py

@@ -0,0 +1,27 @@
+"""python scrape_interactive.py URL --output records.jsonl [--channel chrome]"""
+import argparse
+from playwright.sync_api import sync_playwright
+from common import extract,save
+
+def scrape(url,channel=None):
+    with sync_playwright() as p:
+        browser=p.chromium.launch(headless=True,**({'channel':channel} if channel else {}))
+        try:
+            page=browser.new_page()
+            response=page.goto(url)
+            if not response or not response.ok: raise RuntimeError('Публикация недоступна')
+            page.locator('.lab-records').wait_for(timeout=30000)
+            while True:
+                more=page.get_by_role('button',name='Загрузить ещё',exact=True)
+                if not more.is_visible(): break
+                old=page.locator('.record[data-record-id]').count()
+                more.click()
+                page.wait_for_function('(n)=>document.querySelectorAll(".record[data-record-id]").length > n',arg=old,timeout=10000)
+            records=extract(page.content())
+            if len({r['id'] for r in records})!=len(records): raise RuntimeError('Обнаружены дубликаты записей')
+            return records
+        finally: browser.close()
+
+if __name__=='__main__':
+    p=argparse.ArgumentParser();p.add_argument('url');p.add_argument('--output',default='records.jsonl');p.add_argument('--channel');a=p.parse_args()
+    records=scrape(a.url,a.channel);save(records,a.output);print(f'Сохранено записей: {len(records)}')