#!/usr/bin/env python3 """ import_new_content.py Importa los ew4r_content items no-carta nuevos (id > 9043, catid NOT IN 27,40,41) al WordPress local (Docker). Mapping catid → WP term_ids: 54 (Índice multimedia) → 26 77 (Videos) → 58 64 (Noticias de alcance) → 41 52 (Tablón de anuncios) → 1 (uncategorized / sin categoría) 63 (Fechas) → 40 61 (Lista completa de autores) → 38 65 (Cantoral Salomé Arricibita) → 31 otro → 1 (uncategorized) """ import subprocess import sys from datetime import datetime JOOMLA_SSH_HOST = "134.0.10.170" JOOMLA_SSH_USER = "feadulta" JOOMLA_SSH_PASS = "6Rm2qOF@eundwpda" JOOMLA_DB_HOST = "127.0.0.1" JOOMLA_DB_USER = "fejoomla3" JOOMLA_DB_PASS = "5FF-}5^[>7^pK4W9" JOOMLA_DB_NAME = "fejoomla3" WP_DOCKER = "wordpress-mysql" WP_DB_USER = "wordpress_user" WP_DB_PASS = "wordpress_pass" WP_DB_NAME = "wordpress_db" LAST_CONTENT_ID = 9043 CARTA_CATIDS = {27, 40, 41} CATID_TO_WP = { 54: [26], 77: [58], 64: [41], 52: [1], 63: [40], 61: [38], 65: [31], } DRY_RUN = '--dry-run' in sys.argv # ── Helpers ──────────────────────────────────────────────────────────────────── def joomla_query(query: str) -> list[dict]: mysql_cmd = (f"mysql --skip-ssl -h {JOOMLA_DB_HOST} -u {JOOMLA_DB_USER} " f"-p'{JOOMLA_DB_PASS}' {JOOMLA_DB_NAME} " f"--default-character-set=utf8mb4 -B") cmd = ['sshpass', '-p', JOOMLA_SSH_PASS, 'ssh', f'{JOOMLA_SSH_USER}@{JOOMLA_SSH_HOST}', mysql_cmd] result = subprocess.run(cmd, input=query, capture_output=True, text=True, encoding='utf-8') if result.returncode != 0: print(f"[ERR SSH] {result.stderr[:300]}", file=sys.stderr) return [] lines = result.stdout.strip().split('\n') if len(lines) < 2: return [] headers = lines[0].split('\t') return [dict(zip(headers, line.split('\t'))) for line in lines[1:] if line] def wp_mysql(query: str) -> list[dict]: cmd = ['docker', 'exec', WP_DOCKER, 'mysql', '-u', WP_DB_USER, f'-p{WP_DB_PASS}', WP_DB_NAME, '--default-character-set=utf8mb4', '-B', '-e', query] result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8') if result.returncode != 0: return [] lines = result.stdout.strip().split('\n') if len(lines) < 2: return [] headers = lines[0].split('\t') return [dict(zip(headers, line.split('\t'))) for line in lines[1:] if line] def wp_execute(sql: str): if DRY_RUN: print(f" [DRY] {sql[:110]}") return cmd = ['docker', 'exec', WP_DOCKER, 'mysql', '-u', WP_DB_USER, f'-p{WP_DB_PASS}', WP_DB_NAME, '--default-character-set=utf8mb4', '-e', sql] result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: err = result.stderr.replace('mysql: [Warning] Using a password on the command line interface can be insecure.\n', '') if err.strip(): print(f" [ERR] {err.strip()[:200]}", file=sys.stderr) def esc(s: str) -> str: return s.replace('\\', '\\\\').replace("'", "\\'") def unhex(val: str) -> str: if not val or val == 'NULL': return '' try: return bytes.fromhex(val).decode('utf-8', errors='replace') except Exception: return val def main(): print(f"=== Import ew4r_content no-cartas (id > {LAST_CONTENT_ID}) " f"{'[DRY RUN]' if DRY_RUN else '[LIVE]'} ===\n") # Cargar user map user_rows = wp_mysql( "SELECT um.meta_value jid, u.ID wid FROM wp_users u " "JOIN wp_usermeta um ON um.user_id=u.ID " "WHERE um.meta_key='_fgj2wp_old_user_id'" ) user_map = {} for r in user_rows: try: user_map[int(r['jid'])] = int(r['wid']) except ValueError: pass print(f"Usuarios mapeados: {len(user_map)}") # Cargar term_taxonomy_ids all_term_ids = sorted({t for cats in CATID_TO_WP.values() for t in cats} | {1}) rows = wp_mysql( f"SELECT term_id, term_taxonomy_id FROM wp_term_taxonomy " f"WHERE term_id IN ({','.join(map(str,all_term_ids))}) AND taxonomy='category'" ) tt_ids = {int(r['term_id']): int(r['term_taxonomy_id']) for r in rows} print(f"TT IDs: {tt_ids}") # Polylang ES pl_rows = wp_mysql( "SELECT tt.term_taxonomy_id FROM wp_terms t " "JOIN wp_term_taxonomy tt ON tt.term_id=t.term_id " "WHERE tt.taxonomy='language' AND t.slug='es' LIMIT 1" ) pl_es_tt = int(pl_rows[0]['term_taxonomy_id']) if pl_rows else None # IDs ya en WP existing_rows = wp_mysql( f"SELECT meta_value FROM wp_postmeta " f"WHERE meta_key='_fgj2wp_old_content_id' AND meta_value+0 > {LAST_CONTENT_ID}" ) existing_ids = {int(r['meta_value']) for r in existing_rows} print(f"IDs ya importados con id > {LAST_CONTENT_ID}: {len(existing_ids)}") # Obtener items de Joomla print("\nObteniendo items de Joomla...") catids_excl = ','.join(str(c) for c in CARTA_CATIDS) query = ( f"SELECT id, HEX(title) title, HEX(alias) alias, " f"HEX(introtext) introtext, HEX(`fulltext`) fulltext_col, " f"catid, created, created_by " f"FROM ew4r_content " f"WHERE state=1 AND id > {LAST_CONTENT_ID} AND catid NOT IN ({catids_excl}) " f"ORDER BY id;" ) items = joomla_query(query) print(f"Items a importar: {len(items)}") stats = {'ok': 0, 'skip': 0, 'err': 0} for item in items: joomla_id = int(item['id']) catid = int(item['catid']) title = unhex(item.get('title', '')) alias = unhex(item.get('alias', '')) intro = unhex(item.get('introtext', '')) full = unhex(item.get('fulltext_col', '')) created = item.get('created', '') or datetime.now().strftime('%Y-%m-%d %H:%M:%S') created_by = int(item.get('created_by', 0) or 0) if joomla_id in existing_ids: print(f" [SKIP] id={joomla_id} ya existe") stats['skip'] += 1 continue content = intro + ('\n\n' + full if full.strip() else '') # Multimedia/pensamientos/vídeos/cantoral (catid 54/77/65) son contenido # propio de FeAdulta → autor "Fe Adulta" (WP user 890), no el webmaster # que los sube en Joomla. El resto conserva su autor real (noticias, etc.). FEADULTA_AUTHOR = 890 FEADULTA_CATIDS = {54, 77, 65} wp_author = FEADULTA_AUTHOR if catid in FEADULTA_CATIDS else user_map.get(created_by, 1) wp_cats = CATID_TO_WP.get(catid, [1]) print(f" [{joomla_id}] catid={catid} | {title[:50]}") wp_execute( f"INSERT INTO wp_posts " f"(post_author, post_date, post_date_gmt, post_content, post_title, " f"post_excerpt, post_status, comment_status, ping_status, post_name, " f"post_type, post_modified, post_modified_gmt, comment_count, " f"to_ping, pinged, post_content_filtered) VALUES (" f"{wp_author}, '{created}', '{created}', '{esc(content)}', " f"'{esc(title)}', '', 'publish', 'open', 'open', '{esc(alias[:200])}', " f"'post', '{created}', '{created}', 0, '', '', '')" ) if DRY_RUN: stats['ok'] += 1 continue new_id_rows = wp_mysql("SELECT MAX(ID) new_id FROM wp_posts") if not new_id_rows: stats['err'] += 1 continue new_wp_id = int(new_id_rows[0]['new_id']) print(f" → WP ID={new_wp_id}") # Metas wp_execute(f"INSERT INTO wp_postmeta (post_id, meta_key, meta_value) VALUES ({new_wp_id}, '_fgj2wp_old_content_id', '{joomla_id}')") # Categorías for term_id in wp_cats: tt_id = tt_ids.get(term_id) if tt_id: wp_execute( f"INSERT IGNORE INTO wp_term_relationships (object_id, term_taxonomy_id) " f"VALUES ({new_wp_id}, {tt_id})" ) # Polylang ES if pl_es_tt: wp_execute( f"INSERT IGNORE INTO wp_term_relationships (object_id, term_taxonomy_id) " f"VALUES ({new_wp_id}, {pl_es_tt})" ) stats['ok'] += 1 if not DRY_RUN and stats['ok'] > 0: print("\nActualizando counts de categorías...") tt_str = ','.join(str(v) for v in tt_ids.values()) wp_execute( f"UPDATE wp_term_taxonomy tt SET count = (" f"SELECT COUNT(*) FROM wp_term_relationships tr " f"WHERE tr.term_taxonomy_id=tt.term_taxonomy_id" f") WHERE tt.term_taxonomy_id IN ({tt_str})" ) print(f"\n=== Resultado: {stats['ok']} ok, {stats['skip']} skip, {stats['err']} err ===") if __name__ == '__main__': main()