#!/usr/bin/env python3 """Offline export inspection. Python 3 standard library; no network or HTML execution. Usage: python inspect-export.py confluence|slack INPUT_FOLDER NEW_REPORT_FOLDER [SEARCH_TEXT] Copyright 2026 HowCurio. MIT license: use, modify and redistribute with this notice. Provided without warranty. Do not run on files you lack permission to inspect. """ import csv import json import sys from datetime import datetime, timezone from decimal import Decimal, InvalidOperation from html.parser import HTMLParser from pathlib import Path from urllib.parse import urlsplit, unquote def inside(path, root): return path == root or root in path.parents def csv_write(path, rows, fields): with path.open('w', encoding='utf-8-sig', newline='') as f: w = csv.DictWriter(f, fieldnames=fields) w.writeheader() w.writerows(rows) def cell(value): # Reports preserve data as text without spreadsheet formula execution. s = '' if value is None else str(value) return "'" + s if s.lstrip().startswith(('=', '+', '-', '@')) else s class Links(HTMLParser): def __init__(self): super().__init__(convert_charrefs=True) self.ids = set() self.refs = [] self.base = False def handle_starttag(self, tag, attrs): d = dict(attrs) if d.get('id'): self.ids.add(d['id']) if tag == 'a' and d.get('name'): self.ids.add(d['name']) if tag == 'base' and 'href' in d: self.base = True if tag == 'a' and 'href' in d: self.refs.append(('a', d['href'])) if tag in ('img', 'script', 'iframe', 'audio', 'video', 'source') and 'src' in d: self.refs.append((tag, d['src'])) if tag == 'link' and 'href' in d: self.refs.append(('link', d['href'])) def confluence(root, out): docs = {} for p in sorted(root.rglob('*')): if p.is_symlink(): raise ValueError('Symlinks are not supported: ' + str(p.relative_to(root))) if p.is_file() and p.suffix.lower() in ('.htm', '.html'): parser = Links() parser.feed(p.read_text(encoding='utf-8-sig')) docs[p.resolve()] = parser rows = [] for path, doc in docs.items(): for tag, raw in doc.refs: status = '' target = '' try: u = urlsplit(raw.strip()) if doc.base: status = 'base-url-review' elif u.scheme in ('http', 'https') or u.netloc: status = 'remote-not-fetched' elif u.scheme: status = 'special-scheme-review' elif '\\' in unquote(u.path): status = 'backslash-path-review' elif u.path.startswith('/'): status = 'root-relative-review' else: dest = (path.parent / unquote(u.path)).resolve() if u.path else path if not inside(dest, root): status = 'outside-root' else: target = dest.relative_to(root).as_posix() if not dest.is_file(): status = 'missing-local-file' elif u.fragment and dest in docs and unquote(u.fragment) not in docs[dest].ids: status = 'missing-html-anchor' elif u.fragment and dest not in docs: status = 'non-html-fragment-review' else: status = 'local-present' except (ValueError, OSError) as e: status = 'invalid-reference' rows.append({'page':cell(path.relative_to(root).as_posix()), 'element':tag, 'reference':cell(raw), 'target':cell(target), 'status':status}) csv_write(out/'links.csv', rows, ['page','element','reference','target','status']) counts = {} for r in rows: counts[r['status']] = counts.get(r['status'], 0) + 1 return {'mode':'confluence','htmlPages':len(docs),'references':len(rows),'statusCounts':counts, 'limits':['Static href/src only; no JavaScript, CSS URLs or srcset evaluation.', 'Remote availability, permissions, visible rendering, ZIP safety and source completeness not checked.', 'An HTML base tag makes references manual-review items. No external links fetched.']} def read_json(path): return json.loads(path.read_text(encoding='utf-8-sig')) def utc_ts(value): try: if not isinstance(value, str): return '' n = Decimal(value) if not n.is_finite(): return '' return datetime.fromtimestamp(float(n),timezone.utc).isoformat() except (ValueError, OverflowError, OSError, InvalidOperation): return '' def slack(root, out, query): channels = read_json(root/'channels.json') if not isinstance(channels, list): raise ValueError('channels.json must contain an array') messages, files, summaries, anomalies = [], [], [], [] for c in channels: if not isinstance(c, dict) or not isinstance(c.get('name'), str): raise ValueError('Every channels.json record must have a string name') name = c['name'] folder = (root/name).resolve() if not inside(folder,root) or folder == root or '/' in name or '\\' in name: raise ValueError('Unsafe channel folder name') records = [] dated = sorted(folder.glob('????-??-??.json')) if folder.is_dir() else [] for path in dated: if path.is_symlink(): raise ValueError('Symlinks are not supported') day = read_json(path) if not isinstance(day, list): raise ValueError(str(path.name)+' must contain an array') for index, m in enumerate(day): if not isinstance(m, dict): raise ValueError('Message record must be an object') ts = m.get('ts','') utc = utc_ts(ts) text = m.get('text','') if not isinstance(text, str): raise ValueError('Message text must be a string') previous = m.get('previous',{}) if not isinstance(previous, dict): previous = {} prior = previous.get('text','') if not isinstance(prior, str): prior = '' subtype = str(m.get('subtype','')) source = path.relative_to(root).as_posix() if not utc: anomalies.append({'source':source,'record':index,'issue':'invalid-or-missing-ts','value':ts}) row = {'channel':cell(name),'source':cell(source),'record':index,'ts':cell(ts), 'utc':utc,'user':cell(m.get('user','')),'subtype':cell(subtype), 'thread_ts':cell(m.get('thread_ts','')),'original_ts':cell(m.get('original_ts','')), 'text':cell(text),'previous_text':cell(prior), 'query_match':str(bool(query and query.casefold() in (text+' '+prior).casefold())).lower()} messages.append(row) records.append((str(ts), utc, subtype)) fs = m.get('files',[]) if not isinstance(fs,list): raise ValueError('files must be an array') for f in fs: if not isinstance(f,dict): raise ValueError('File record must be an object') files.append({'channel':cell(name),'source':cell(source),'message_ts':cell(ts), 'id':cell(f.get('id','')),'name':cell(f.get('name',f.get('title',''))), 'permalink':cell(f.get('permalink','')), 'url_private':cell(f.get('url_private','')), 'url_private_download':cell(f.get('url_private_download','')), 'status':'reference-only-not-downloaded'}) valid = sorted((r for r in records if r[1]), key=lambda r:Decimal(r[0])) summaries.append({'channel':name,'datedFiles':len(dated),'records':len(records), 'firstRecordUTC':valid[0][1] if valid else None, 'lastRecordUTC':valid[-1][1] if valid else None, 'edits':sum(r[2]=='message_changed' for r in records), 'deletions':sum(r[2]=='message_deleted' for r in records), 'folderState':'present' if folder.is_dir() else 'absent-not-proof-of-data-loss'}) csv_write(out/'messages.csv',messages,['channel','source','record','ts','utc','user','subtype','thread_ts','original_ts','text','previous_text','query_match']) csv_write(out/'file-references.csv',files,['channel','source','message_ts','id','name','permalink','url_private','url_private_download','status']) return {'mode':'slack','channels':summaries,'messageRecords':len(messages),'fileReferenceOccurrences':len(files), 'uniqueNonemptyFileIds':len(set(f['id'] for f in files if f['id'])), 'query':query,'queryMatches':sum(m['query_match']=='true' for m in messages),'anomalies':anomalies, 'limits':['Only channels.json and direct public-channel dated JSON files are read.', 'Raw records are retained; edit/delete records are not replayed into a final conversation.', 'File reference rows are occurrences, not downloaded files. No remote requests.', 'Plain text search omits block-only text, attachments, canvases, private conversations and deleted unavailable data.']} def main(): if len(sys.argv) not in (4,5) or sys.argv[1] not in ('confluence','slack'): raise ValueError(__doc__.splitlines()[1]) mode, root, out = sys.argv[1],Path(sys.argv[2]).resolve(),Path(sys.argv[3]).resolve() if not root.is_dir(): raise ValueError('INPUT_FOLDER must exist') if inside(out,root): raise ValueError('REPORT_FOLDER must be outside INPUT_FOLDER') if root.is_symlink(): raise ValueError('Input symlinks are not supported') out.mkdir(parents=True,exist_ok=False) result = confluence(root,out) if mode=='confluence' else slack(root,out,sys.argv[4] if len(sys.argv)==5 else '') (out/'summary.json').write_text(json.dumps(result,ensure_ascii=False,indent=2)+'\n',encoding='utf-8') print(json.dumps(result,ensure_ascii=False,indent=2)) if __name__ == '__main__': try: main() except (ValueError,OSError,json.JSONDecodeError) as e: print('Inspection stopped: '+str(e),file=sys.stderr) sys.exit(1)