feat: add backfill elempleo_co (checkpoint reanudable)

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()