From bb5afa5f4299c166bfaf8fbcde65b40dcda575e3 Mon Sep 17 00:00:00 2001 From: claude Date: Sun, 2 Aug 2026 04:23:10 +0200 Subject: [PATCH] feat: add backfill elempleo_co (checkpoint reanudable) --- inmuebles/elempleo_co/backfill.py | 196 ++++++++++++++++++++++++++++++ 1 file changed, 196 insertions(+) create mode 100644 inmuebles/elempleo_co/backfill.py diff --git a/inmuebles/elempleo_co/backfill.py b/inmuebles/elempleo_co/backfill.py new file mode 100644 index 0000000..28c1cb3 --- /dev/null +++ b/inmuebles/elempleo_co/backfill.py @@ -0,0 +1,196 @@ +#!/usr/bin/env python3 +""" +Backfill scraper ElEmpleo.com Colombia +Checkpoint: /opt/scrapers/elempleo_co/backfill_state.json + {"page": N, "total_new": M} -> reanudar desde pagina N +Para reiniciar desde cero: borrar el archivo de checkpoint. +""" +import os, re, time, logging, json +import requests +import psycopg2 +from psycopg2.extras import execute_values +from datetime import datetime + +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s %(levelname)s %(message)s', + datefmt='%Y-%m-%d %H:%M:%S', +) +log = logging.getLogger('elempleo_backfill') + +PROXY = 'socks5h://127.0.0.1:1090' +PROXIES = {'http': PROXY, 'https': PROXY} +DB_HOST = '100.75.240.87' +DB_NAME = 'elempleo_co' +DB_USER = 'pgadmin' +DB_PASS = 'J5BVlq65JvedWxZVrcY96OQX' +UA = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/124 Safari/537.36' +BASE = 'https://www.elempleo.com' +LIST_URL = BASE + '/co/ofertas-empleo/?pagina={}' +DELAY = 0.8 +MAX_PAGES = int(os.environ.get('BF_MAX_PAGES', '500')) +MAX_STREAK = int(os.environ.get('BF_STREAK', '8')) + +STATE_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'backfill_state.json') + + +def load_state(): + if os.path.exists(STATE_FILE): + with open(STATE_FILE) as f: + st = json.load(f) + log.info('Checkpoint cargado: pagina=%d total_new=%d', st['page'], st.get('total_new', 0)) + return st + return {'page': 1, 'total_new': 0} + + +def save_state(page, total_new): + with open(STATE_FILE, 'w') as f: + json.dump({'page': page, 'total_new': total_new, 'ts': datetime.utcnow().isoformat()}, f) + + +def get_session(): + s = requests.Session() + s.proxies.update(PROXIES) + s.headers.update({'User-Agent': UA, 'Accept': 'text/html,application/xhtml+xml,*/*;q=0.8', + 'Accept-Language': 'es-CO,es;q=0.9'}) + return s + + +def get_conn(): + return psycopg2.connect(host=DB_HOST, port=5432, dbname=DB_NAME, + user=DB_USER, password=DB_PASS, connect_timeout=10) + + +def create_table(conn): + cur = conn.cursor() + cur.execute(""" + CREATE TABLE IF NOT EXISTS jobs ( + job_id TEXT PRIMARY KEY, + titulo TEXT, + empresa TEXT, + ubicacion TEXT, + url TEXT UNIQUE, + fecha_pub TEXT, + categoria TEXT, + scraped_at TIMESTAMP DEFAULT NOW() + ) + """) + conn.commit() + cur.close() + + +def upsert(conn, rows): + if not rows: + return 0 + deduped = list({r['job_id']: r for r in rows}.values()) + cur = conn.cursor() + execute_values(cur, + """INSERT INTO jobs (job_id, titulo, empresa, ubicacion, url, scraped_at) + VALUES %s + ON CONFLICT (job_id) DO UPDATE SET + titulo=EXCLUDED.titulo, scraped_at=EXCLUDED.scraped_at""", + [(r['job_id'], r['titulo'], r['empresa'], r['ubicacion'], + r['url'], datetime.utcnow()) for r in deduped] + ) + n = cur.rowcount + conn.commit() + cur.close() + return n + + +def fetch_page(sess, page): + url = LIST_URL.format(page) + for attempt in range(3): + try: + resp = sess.get(url, timeout=30) + if resp.status_code == 404: + return None + if resp.status_code in (403, 429): + log.warning('HTTP %d en %s', resp.status_code, url) + time.sleep(60) + continue + resp.raise_for_status() + break + except Exception as e: + if attempt == 2: + log.error('Error %s: %s', url, e) + return [] + time.sleep([2, 5, 15][attempt]) + + html = resp.text + links = list(dict.fromkeys(re.findall(r'href="(/co/ofertas-trabajo/[^"]{10,150})"', html))) + if not links: + return [] + + jobs = [] + for href in links: + m = re.search(r'-(\d{7,})$', href) + if not m: + continue + job_id = m.group(1) + slug = href.split('/')[-1] + titulo_raw = slug[:-(len(job_id) + 1)].replace('-', ' ').strip() + titulo = titulo_raw.title() if titulo_raw else 'Sin titulo' + jobs.append({ + 'job_id': job_id, + 'titulo': titulo[:300], + 'empresa': '', + 'ubicacion': 'Colombia', + 'url': BASE + href, + }) + + log.info('Pagina %d: %d ofertas', page, len(jobs)) + return jobs + + +def main(): + state = load_state() + start_page = state['page'] + total_new = state.get('total_new', 0) + log.info('=== Backfill ElEmpleo CO — desde pagina %d ===', start_page) + + sess = get_session() + conn = get_conn() + create_table(conn) + + streak = 0 + last_page = start_page + + for page in range(start_page, MAX_PAGES + 1): + last_page = page + jobs = fetch_page(sess, page) + + if jobs is None: + log.info('Pagina %d: 404 — fin de paginacion', page) + save_state(page + 1, total_new) + break + + if not jobs: + streak += 1 + log.info('Pagina %d: sin resultados (racha %d/%d)', page, streak, MAX_STREAK) + save_state(page + 1, total_new) + if streak >= MAX_STREAK: + log.info('Deteniendo: %d paginas consecutivas sin resultados', streak) + break + time.sleep(DELAY) + continue + + n = upsert(conn, jobs) + streak = 0 if n > 0 else streak + 1 + total_new += n + save_state(page + 1, total_new) + log.info('Pagina %d: %d nuevos (acumulado %d)', page, n, total_new) + + if streak >= MAX_STREAK: + log.info('Deteniendo: %d paginas sin nuevos', streak) + break + + time.sleep(DELAY) + + conn.close() + log.info('=== Backfill finalizado: %d nuevos (ultima pagina: %d) ===', total_new, last_page) + print(f'Finalizado: {total_new} nuevos') + + +if __name__ == '__main__': + main()