255 lines
9.0 KiB
Python
255 lines
9.0 KiB
Python
#!/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<!--more-->\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()
|