Syncy is a script, made with Claude, which synchronizes the ๐ posts made online (typically in real time) to my local laussywiki (llw), so as to keep everything harmonious in one and the same branch of the multiverse.
It should be patched so that the cloudflare secret key is stored in the credential files, like llw2lw (maybe even using the same file as those are the same credentials).
# ___
#/ __|_ _ _ _ __ _| _|
#\__ \ || | ' \/ _| _| _|
#|___/\_, |_||_\__| _|
# |__/ SyncY _|
# V1.0 Laussy & Claude _|
# Synchronize Y posts from lw to llw
#
#!/usr/bin/env python3
"""
syncy "June (2026)"
Pull the named Y month-page from the ONLINE wiki into the LOCAL wiki:
fetch both, merge by day/time (latest edit wins), and write the result back to
the LOCAL wiki only โ local is the base of record. Online is never modified.
It writes only when online actually contributed something (a post local lacks,
or a newer edit of a post local has). If local already contains everything
online has, nothing is written. A backup of the local page is saved first.
Setup:
- Credentials: a MediaWiki bot password from the LOCAL wiki's
Special:BotPasswords. Grants needed: "Edit existing pages" AND "Upload,
replace, and move files" (uploads need their own grant). Provide via env
vars SYNCY_USER / SYNCY_PASS or a ~/.syncy file (chmod 600):
user = Fabrice@syncy
pass = xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
- Local must have $wgEnableUploads = true (it does, since the gadget uploads).
"""
import sys, os, re, html, json
import urllib.request, urllib.parse, http.cookiejar
# ---- config ---------------------------------------------------------------
LOCAL_BASE = 'http://localhost/laussywiki/index.php/' # article path (read)
LOCAL_API = 'http://localhost/laussywiki/api.php' # API (write)
ONLINE_BASE = 'https://laussy.org/wiki/' # article path (read)
USER_AGENT = 'Mozilla/5.0 (Y-sync)'
# Secret header to get the ONLINE read past Cloudflare (matches your WAF rule).
EXTRA_HEADERS = {
'X-Y-Sync': '---replace-with-secret-key---',
}
BACKUP_DIR = os.path.expanduser('~/.syncy-backups')
# ---------------------------------------------------------------------------
MONTHS = ['January', 'February', 'March', 'April', 'May', 'June',
'July', 'August', 'September', 'October', 'November', 'December']
POST_RE = re.compile(r'<blockquote id="(Y-\d{8}-\d{6})">(.*?)</blockquote>', re.S)
TEXT_RE = re.compile(r'<text\b[^>]*>(.*?)</text>', re.S)
EDIT_RE = re.compile(r'<!--lastedited:(\d{8}-\d{6})-->')
FILE_RE = re.compile(r'\[\[\s*(?:File|Image)\s*:\s*([^|\]]+?)\s*(?:\|[^\]]*)?\]\]', re.I)
# ---- read (Special:Export, no auth needed for public reads) ---------------
def fetch(base, title):
url = base + 'Special:Export/' + urllib.parse.quote(title.replace(' ', '_'))
headers = {'User-Agent': USER_AGENT}
headers.update(EXTRA_HEADERS)
req = urllib.request.Request(url, headers=headers)
with urllib.request.urlopen(req, timeout=30) as r:
xml = r.read().decode('utf-8', 'replace')
m = TEXT_RE.findall(xml)
return html.unescape(m[-1]) if m else '' # '' if the page doesn't exist
# ---- merge ----------------------------------------------------------------
def last_changed(pid, inner):
m = EDIT_RE.findall(inner)
return max(m) if m else pid[2:] # 'YYYYMMDD-HHMMSS'
def collect(texts):
kept, clashes = {}, {}
for t in texts:
for m in POST_RE.finditer(t):
pid, inner = m.group(1), m.group(2)
lc = last_changed(pid, inner)
if pid not in kept:
kept[pid] = (lc, inner)
else:
old_lc, old_inner = kept[pid]
if lc > old_lc:
kept[pid] = (lc, inner)
elif lc == old_lc and old_inner.strip() != inner.strip():
clashes.setdefault(pid, [old_inner]).append(inner)
return kept, clashes
def build(kept, clashes):
posts = {pid: inner for pid, (lc, inner) in kept.items()}
for pid, variants in clashes.items():
for i, v in enumerate(variants[1:]):
nid = pid + chr(ord('b') + i)
posts[nid] = v
sys.stderr.write('CLASH: %s differs with the same last-edit time; '
'kept both (extra id %s).\n' % (pid, nid))
days = {}
for pid in posts:
days.setdefault(pid[2:10], []).append(pid)
out = []
for day in sorted(days):
y, mo, d = int(day[0:4]), int(day[4:6]), int(day[6:8])
out.append('== {{thisday|%d|%s|%d}} ==' % (d, MONTHS[mo - 1], y))
first = True
for pid in sorted(days[day]):
if not first:
out.append('-----')
first = False
out.append('<blockquote id="%s">%s</blockquote>' % (pid, posts[pid]))
out.append('')
return '\n'.join(out).rstrip() + '\n'
def normalize(kept):
"""Post-level view for change detection (ignores whitespace/formatting)."""
return {pid: inner.strip() for pid, (lc, inner) in kept.items()}
# ---- LOCAL API session (login once; reuse for query, upload, edit) --------
def load_creds():
u, p = os.environ.get('SYNCY_USER'), os.environ.get('SYNCY_PASS')
if u and p:
return u, p
path = os.path.expanduser('~/.syncy')
if os.path.exists(path):
d = {}
for line in open(path, encoding='utf-8'):
line = line.strip()
if line and not line.startswith('#') and '=' in line:
k, v = line.split('=', 1)
d[k.strip().lower()] = v.strip()
if d.get('user') and d.get('pass'):
return d['user'], d['pass']
sys.exit('No credentials: set SYNCY_USER/SYNCY_PASS or create ~/.syncy '
'(user=..., pass=...).')
def make_api():
cj = http.cookiejar.CookieJar()
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj))
def call(params, post=False):
params = dict(params); params['format'] = 'json'
if post:
req = urllib.request.Request(
LOCAL_API, data=urllib.parse.urlencode(params).encode('utf-8'),
headers={'User-Agent': USER_AGENT})
else:
req = urllib.request.Request(
LOCAL_API + '?' + urllib.parse.urlencode(params),
headers={'User-Agent': USER_AGENT})
with opener.open(req, timeout=30) as r:
return json.loads(r.read().decode('utf-8'))
return call, opener
def login(call):
user, pwd = load_creds()
lt = call({'action': 'query', 'meta': 'tokens', 'type': 'login'})
lt = lt['query']['tokens']['logintoken']
r = call({'action': 'login', 'lgname': user, 'lgpassword': pwd, 'lgtoken': lt}, post=True)
if r.get('login', {}).get('result') != 'Success':
sys.exit('LOGIN FAILED: ' + json.dumps(r.get('login', r)))
def csrf_token(call):
return call({'action': 'query', 'meta': 'tokens'})['query']['tokens']['csrftoken']
def edit_page(call, csrf, title, text, summary):
r = call({'action': 'edit', 'title': title, 'text': text,
'summary': summary, 'token': csrf, 'assert': 'user'}, post=True)
if 'error' in r:
sys.exit('EDIT FAILED: ' + json.dumps(r['error']))
return r['edit']
# ---- files: find references, see what's missing locally, transfer ---------
def referenced_files(text):
names = []
for m in FILE_RE.finditer(text):
n = m.group(1).strip().replace(' ', '_')
if n and n not in names:
names.append(n)
return names
def local_missing(call, names):
"""Read-only (no login): which of these File: names lack a file locally."""
missing = []
for i in range(0, len(names), 50):
chunk = names[i:i + 50]
r = call({'action': 'query', 'prop': 'imageinfo', 'iiprop': 'timestamp',
'titles': '|'.join('File:' + n for n in chunk), 'formatversion': '2'})
present = set()
for pg in r.get('query', {}).get('pages', []):
if 'imageinfo' in pg: # a file actually exists for this title
present.add(pg['title'])
for n in chunk:
if ('File:' + n.replace('_', ' ')) not in present:
missing.append(n)
return missing
def download_online_file(name):
"""Exact bytes of a file from the online wiki via Special:FilePath (which
302-redirects to the real image; urllib carries our headers across it)."""
url = ONLINE_BASE + 'Special:FilePath/' + urllib.parse.quote(name)
headers = {'User-Agent': USER_AGENT}
headers.update(EXTRA_HEADERS)
req = urllib.request.Request(url, headers=headers)
with urllib.request.urlopen(req, timeout=120) as r:
return r.read()
def upload_local(opener, csrf, name, data, comment):
boundary = '----syncy' + os.urandom(8).hex()
def field(n, v):
return ('--%s\r\nContent-Disposition: form-data; name="%s"\r\n\r\n%s\r\n'
% (boundary, n, v)).encode('utf-8')
head = b''.join([field('action', 'upload'), field('filename', name),
field('token', csrf), field('ignorewarnings', '1'),
field('comment', comment), field('format', 'json')])
filehdr = ('--%s\r\nContent-Disposition: form-data; name="file"; filename="%s"\r\n'
'Content-Type: application/octet-stream\r\n\r\n' % (boundary, name)).encode('utf-8')
body = head + filehdr + data + b'\r\n' + ('--%s--\r\n' % boundary).encode('utf-8')
req = urllib.request.Request(LOCAL_API, data=body, headers={
'User-Agent': USER_AGENT,
'Content-Type': 'multipart/form-data; boundary=%s' % boundary})
with opener.open(req, timeout=300) as r:
return json.loads(r.read().decode('utf-8'))
def transfer_files(opener, csrf, names):
done, failed = [], []
for n in names:
try:
data = download_online_file(n)
r = upload_local(opener, csrf, n, data, 'syncy: import file from online')
if r.get('upload', {}).get('result') in ('Success', 'Warning'):
done.append(n)
else:
failed.append((n, json.dumps(r)[:200]))
except Exception as e:
failed.append((n, str(e)))
return done, failed
def backup(title, text):
os.makedirs(BACKUP_DIR, exist_ok=True)
from datetime import datetime
name = '%s.%s.txt' % (title.replace(' ', '_').replace('/', '_'),
datetime.now().strftime('%Y%m%d-%H%M%S'))
path = os.path.join(BACKUP_DIR, name)
with open(path, 'w', encoding='utf-8') as f:
f.write(text)
return path
# ---- main -----------------------------------------------------------------
def main():
if len(sys.argv) != 2:
sys.stderr.write(__doc__)
sys.exit(1)
title = sys.argv[1]
try:
local_text = fetch(LOCAL_BASE, title)
except Exception as e:
sys.exit('ERROR reading local "%s": %s' % (title, e))
try:
online_text = fetch(ONLINE_BASE, title)
except Exception as e:
sys.exit('ERROR reading online "%s": %s (Cloudflare? check EXTRA_HEADERS)' % (title, e))
both, clashes = collect([local_text, online_text])
local_only, _ = collect([local_text])
merged = build(both, clashes)
textchanged = bool(clashes) or normalize(both) != normalize(local_only)
new_ids = sorted(set(both) - set(local_only))
upd_ids = sorted(pid for pid in both if pid in local_only
and both[pid][1].strip() != local_only[pid][1].strip())
refs = referenced_files(merged)
call, opener = make_api()
missing = local_missing(call, refs) # read-only, no login required
if not textchanged and not missing:
print('Local already has every post and file from online; nothing to do.')
return
login(call) # uploads/edits need auth
csrf = csrf_token(call)
done, failed = transfer_files(opener, csrf, missing) if missing else ([], [])
rev = bpath = None
if textchanged:
bpath = backup(title, local_text)
rev = edit_page(call, csrf, title, merged, 'syncy: import online posts').get('newrevid')
print('Done with "%s".' % title)
if textchanged:
print(' posts: %d new, %d updated (local rev %s; backup %s)'
% (len(new_ids), len(upd_ids), rev, bpath))
else:
print(' posts: no text changes')
print(' files: %d transferred, %d already present, %d failed'
% (len(done), len(refs) - len(missing), len(failed)))
for n, why in failed:
print(' FAILED %s: %s' % (n, why))
if __name__ == '__main__':
main()