From f75deade23efda513401cf13609c62caef82c618 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Sat, 20 Jun 2026 13:45:52 +1000 Subject: [PATCH] feat: add django export tooling --- scripts/README.md | 36 ++ .../export-django-links.cpython-312.pyc | Bin 0 -> 17062 bytes scripts/export-django-links.py | 418 ++++++++++++++++++ scripts/export-django-links.test.py | 340 ++++++++++++++ 4 files changed, 794 insertions(+) create mode 100644 scripts/README.md create mode 100644 scripts/__pycache__/export-django-links.cpython-312.pyc create mode 100644 scripts/export-django-links.py create mode 100644 scripts/export-django-links.test.py diff --git a/scripts/README.md b/scripts/README.md new file mode 100644 index 0000000..6fa2f9c --- /dev/null +++ b/scripts/README.md @@ -0,0 +1,36 @@ +# Django Links Export Tool + +Standalone Python script to export shortlinks from the old Linux/Django +`links` app SQLite database into JSONL format suitable for import into +the Cloudflare D1 shortlinks system. + +## Usage + +```bash +# Export from a Django SQLite DB to stdout +python3 scripts/export-django-links.py --db /path/to/db.sqlite3 + +# Export to file +python3 scripts/export-django-links.py --db /path/to/db.sqlite3 --output ./export.jsonl + +# Use env var instead of --db +export DJANGO_DB_PATH=/path/to/db.sqlite3 +python3 scripts/export-django-links.py --output ./export.jsonl + +# Filter by owner email (only when links_link has an owner FK column) +python3 scripts/export-django-links.py --db /path/to/db.sqlite3 --owner-email admin@example.com +``` + +## Output format + +Line-delimited JSON (JSONL). Each line: + +```json +{"alias":"claude","link_type":"redirect","target_url":"https://claude.ai","content_markdown":"","description":"","tags":["ai","tools"],"created_at":"...","updated_at":"..."} +``` + +## Running tests + +```bash +python3 scripts/export-django-links.test.py +``` diff --git a/scripts/__pycache__/export-django-links.cpython-312.pyc b/scripts/__pycache__/export-django-links.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..eaffd456e72320c0e54284fddfb8e6ed65fc7a56 GIT binary patch literal 17062 zcmcJ0dvF`andb~J00wV>Z&IW*q9~J)Ny@Ti>n(>8DT#VeK1sGLIbIUtfFvjopl3i) zguyKAxJsm=I#F?QqGNl_taoeai}zNkTwQcksXAM?xytQc)qy4xQ4>|2u6C<#bAQpH zcGpqrs_ysoJODD3-T#&Tn~KD4gTmkj{7_MVP1k>;g?_L zIqoJWbLTmkmu(4t$adbwQ{A4hpSQESa9&_{$9V_4JI_1W-F4oDyO0ov+~?gxp7S1z zbI4BF^`7m#S9YBD$>Mpx>^>im*U6r@g!4hUR`%i^lIvt2?qPYo?8m)C-XI5XFO?-Z zhl#*kKW@$Jv zf~M$5dZ2qmjVV1IdUk3wiB-ggqVWV~P0CV_%vRCen@Xl*$#i#mY&h1VFEe@}5tHP& z+MBu*Q^q0;d{n{XRS&(>(bFfpq|}AMSZ`Vy#-vzEOlqRm;}K~zo*rN?+8a%#lJVYX zqVRrFUp$tO)!iPCge#heN7ZH;eUUx9SxQHh{#d$uM8Qj#%NmL*7iBEIS(0OFuM!_l z$5Y8>HbFYtkMX@qESiqV-O+TjG%}10WOa|{%s^a~^brzjarTxN8%-p}B-J{h-BLU) z#fRvWR4JKC55$xGQlF93C>l_)mK_T1ao2v)nwCWTVhjsoW1D)|V!PvV4-UMC)@;_z{kkwlEsXY?rO|;{QW{obDmI{5 z8cfB3v>=;wip^HUY9uu!^|Z9G*;^R1>A~wHLHG$U!B*)kO$548M|uZ}|<0 zP3*)fXlpdl#CXlAqdPzZq7cn#kceFq%}vr=#tv!z726XLHCtTP9E{vFw?$Bzub65y zAr11-yhS*!dCYmDY0Xo>zFtF^&^vG>vh5tlFYiX>CYQGQvRv9m0CWu+f?fZmmml*# zF#36JjEmU1upI*No`_v@upQOx>X@oI^_@hoeMs$BX@2R^b6W>eL$R&Vc*})UdTTEv z%GQC{Sbu7(j(X}=U81$<(gRopvDG>}rn$QH*_2vL(}q7agd)e?7u*w@zH{*A(Hlp9 zA#7Mi8@KjbPUHDExh!{?S2yrnngH1~XuYf1g13g0U46;?4~!ie=d(O$!ev6pHnRuw zy3N)>rYxDi^t)^?g0#8@eOA?~3dQjD=7Xdz%@1G=1SX^}55?v#`o+K;w%9X{FJexUn>wlhas zrH;#B6RD&$6jxP{Z>ttq36a?YQcfwmi6@woLeO4{%Q3lCdQpu@G`j>orpCZ}SI+(` z!i-FfF$0D(`3kBB;~PoJyQR#Ng+8`k=mAl%Sdw%pWvrRxJsN1(ZYko{Y$;W9#gdoe zN-C+@p^7kMR;utAO#qupYXVK9Idzt$IpQiN1Ltr;ap{&XXfF9eH$kYhGD%X_(a0e9 zChZ&mSf{cDtsVGNuc63s4+3SM1$Mq?`@r*#=YzmIfqdhR`Kq0{!}q);_r+j7bYem5 zx-WXKi<9D1_?q{=*MEJ-}+Q~^xT5DqZriXzB2m8=#}v| z#;49Mxat__zT5q`&*ejh7sMk)po`v;T$fUZ1+7h;{}Vkh8~7~uE+^aGw~zDgB962c zjb>p|d@PupXxvsr(N0XZ>dM->El!rT4SH8~&}P}NC=G^ZqRR10Ad3X@^cLK&n&W~)$(-^LaeX(@!0JM*Ypgcnj4hF9YI3QKy z6_xN3Ctw_bU!P6adsJmFIxpZ)-Gd^>VJolio!pzRX_<}9p1pM-U$tZ2yYsHMZO+>^ z?>(4nU)-?i7v6^HjT3DXec$T*%-fI`8h(Qtd&;bDhp>WliOASi?qq`1{n*Wfcr=dS zOSci5uJ}RJ0&N_&A0kC|>h<@reUCXRmvuyz;Kgv`T6sjTZlt40Tcn@ma7-dS!Cc?C zAiJ}|B~In9RmvWtR%=4h0iNarF6}qxm%ZlG-{iqAEG99y1_J`SaKN7d#s)@k$zlY) z%%MYZXqu#aGhK%%*oJ<6DGq7!yH<3w6iYyIMWL+^KnKPxnrwY^O1JXJV((5wFOZ?m z>~-nTsgozxnnpT%q~laa0bbfCJukI&wHw3m3zEmd7NrSv>(boThrD7))g!r2Gcq(;qWSFtFj~qJr*ue9gYz?3uNLx^fsuE8i8ZwBe)O1X+ zljvs9$}zz7<4^rhD01AA;Bu5K`oq`1KKb<}8}EPWfv;}aB?LVGEOOq`$);~@S%NRf z`+0y9J&Pq}Q?K0Enh)2{Y@9hTGy45wGyC$vZFzCq1AT%;anG{d7I6L#XW)&2Zw@Xw zIEQ!1$py-Y)zwY#_k*?5XJ)p|y!8ED^TaJ7N~h$j&rg5jR@rA8o?R?0n|kTSuFrh6 z#02XeVgVKTibM0B!+GKG7Z0n87$?uxeATvOM@{FJZ%PMv?hapmpxS+Q?gir1UhHdb`S)PYxDc(n>u&xodD*J(d5u5TnZkionT}m6O$^j}4 zqIkqhna)3$AQ{;+`aRxxtV3$lInic=P-(juwHkz{v{QpcbDB$u4JV?#u>$cahXApf zuu3e#!nzH*dI50XAlN*LHR-DM52UNR!CiB~UAH&gR_B8U^Ws5^$WH%(L>8Vpb9Mi; zJR^TEF*|x&{%q@kRiw5ga=y@4p)enJ$5(!!!G32GkFvK|mmnLG3DJK-<%;-wH4>X_ z@B3=v6go&d_i7jLUC0W$P#t%qExQ(&Z&B7&v0zP3h<&{5cuSO>pi@@{tLn40b0!s; zls4{MHA2J)ch>eLBV1W0=%+L5LU}yvu;|AV@n-g{q69)e(}So2Q~D+H)TIZM)JQ*^ zzvN(HbP-LCwPsAY_tnJ@!}g{Ex<4 zv;6eN>GXS%x#|DHIgM!*((RQOQLK0cF~7tUiDnn(cdWN&`n~xA4~3n=1Zz@fwyRO9DBsA3v>^AJ%$mQ+hQwl zf5CefCttm)bkb;2I5aODsk%__yQq2S2aV{vsGkRkAJPKo`kHQ87dV!66p9S6@8VDW z6BO8aNa~V`o3%G;p$E91eBi5?*gK`<-8Iuk^5SOjt`aA70#}epb~on;E_t{>)vr0x zQ8mFY218R7lVcNtF5pOJU5tR07`gyXLlK{`4_{9tXlc zUT-&v{Y$lWX8gR06B>ufleMoU#7*XL2^z3z7$pVNn5O?_LOaU?%0c3pG`efs1&*9W zUqRru@rdX>#$vD>WncbY$-1xb92buj^}@8Lztfy4+@axLGwU$*_ne3hOIEn76&`rR z4Vd#9v4+%WOzA~LPL42lI&%pd?rd^sWi}YXJT61+r|6olVq4Q?>Wcu#ja=Y^7L4X# zBQ??NS4gBsoJEvXkmkb07<{)5wlK@=q=BkJj$(z3$yIizCy|QEEEcFL7t#F@f&2so ztpD2exsD~9%Tc`;EWMtc%;szN%?J124IZ8g9-a?&P6$f@E>QAoj(5DwgCd%?-rDx# z=YIIyk9N*A?EynO{Jkxogg@E+Y2DrG7w4*9oUcAR;h2a{36uR-1Ej~k%s=pka>u`T z7%ozD9ggZHo8YbnJ>1-TV{cw+pD*pWTiP{O+BIML!esjwe;=&48*H2lHsZMiR(0(- zCUaLm`rN_!!%GfK@JM|YFYwlewp#Af+R_70**|@1!-0_PPLN0aPN@9A7RQ~ zs&BDVeJk~@@Hh}BW)hxMw3S|V_)-ns7TznP#&M9^9wMi#Jqt+!X(?tgqi%YR@ex~Q z_o-NVMA3amI=UP>X2v?$;lm$lnv<)9B?1+?apXUFwyWb*_ra6LUp&!u+BjyV8%t!& zP&td*szWZ7V`=z|y7i!i?qg7rfcy6Z+lnH`{k^wpx@MU#^9W(2fhV=_)vO;O7$!KOfYEaWN}qT?X2T@b!5G zZrqJlL8%5pqDLvm-F=By!ddrV@w%`qy0U^PWZ@cuZoiiJ%(_h&IK>!($&U9O1XS-sx|$!W%yW(t7n-i%V1Z4id*-b(wwRe(2_PMSJ2jp9t~hU2EO4<#yw_^Wb`;gdK277}MC3489af79+4h%oDC0U;-H7uX9II4S>?c>4m0Q}40%z*m;!#=dRe zQ<7&yJU63ut}BH?J_qNdAL2^UD${mQ4x%HO|YX zJt@MT>02u%K>_t=EG} zHMakT$^JKe0Y2l9I=iH%M|=c3uX}RQL|)=dDa_{fj#JWsb9z3I)OPw{#G{8ane6dV zzJsY_Aci@N0FkT-l-{8^)HHIflsFcEFlRb8gmeL9F#v!mrIAu5sC12_hu-fsf^=KT zhvwC*OtDo434*L@CcZSU8cUF>LhfBnpnNyYixd+jhWIGburw!bQ9LnFG)H%$7vVD) z%C^{$*+E(_Y*1D6G5^*8%cfD0-0)gLCE8>Xa%G_#j^3heS11bmU-75D3!^K?eO|_i zrBh@1$|i(MmuxowriXUH?RgmDeANZzY5kU8l1FA+sEdfJiI8D-W6-+#G2_{3*zPn{<5hvH_zQTH~sp2`IGsw=Gnuyj^5gP z`?b8c4YIgyL+;pORZZ^D=T2h{&5P9ycdNI|Rd1QEZoXUn^j!7R^VQoY0{4Pt_kE=> zX~XUOL%ZD{fQa^2-Sw@X^R1sA`MGcNgKf|Kc-;@z{iyD4-P!N=&-Bhd{Ym}L0`mu6 z`d)9o?(9U@T;17wr5o>uDi$cAYx6g^&C+v%YEAN%&1SxmW2!Ku7r(c@a zgImAu;JMOWzgceJ*1ycF6*#0nb9Y39eoX1Y(vXBy%c0hK-Pux)5D*5$<$8nE z(UVz6t>hvzXx*9-74(augxK>IC0ox`iY0Is4G0>DNOQf zMUKrgBUi^HFZ1NONP>TlJ7n(!&Re#<3#Bw8 zWo}*bggW)iTDd(?~lcpXOAU4X=>N3+tQSQS&o#dFI>;9o;9|UeLnr9f#UpJbtFzYG!l+ z3CjrWZ46O*s31a9{v{R6MkKDJ`IYFX5x+#-Re!mg=uiojr5;me>BQ$>v4hI{0I8w9 z`!iIqcW}_uN;A*=#m?{V{BZZ&x|WIFsfMfZTibJo|54x_J0MHSE2b{K-IY7B=qh{9 zKj+$fzii`7%Us!0lkR)|n)_v&C*2RqHcq%7oH&y|(Q|F@v@qR2Gj=O-`;`T6XTEPB z*N!<#N~d*gyP7ei&#<9j=0+&&Mxea_jXoOcn zSPh}(s>@_aFq)7>5%FZ&&r+s6a*c>w;B8P@zWs5^d`*p{Sq2%)hfB-&*(&MICKVd| z2$GX2OCL%awAU7mAI&&M(tRyEk$40*15!Dmw=1-`L!eh5a??kkaYfbMcpQmqvA*a? zBK?T>{0c`xw8XOBC`A;H(F95!(j4p~0BDFR^5YyVYgw5FwC3ndq|{i%L7_&rY~;=w zYt?Lh1L&vKfjz`}S4SO%*ibyJ{1CJ3C7d2Z1s;k(`1;`FVD7*@S8(dcjQ!gu@B713 zP16Tw8yEafPuLcH{#?gmu!4LNrO+Wn4`s)Fr|f3UjT!*OesX#t-1bTPT=>X>*tsbB zr^1t-NeB71N+EVc&v!}*a?Z8R7!^uvd7I7C`IQq3A5wl~k9>5wiJr>q0`!K72<*Y!c0CJhuNZHM# zbyKeJafYgAb5g!|;S&`-(%eV}pz#p{<-q|J+1DPeQsGONJ(8Sa-$bPKk0QpHiMI0F ziB{m9fb`CO#(*;d<#uWUWr}9@zd-qL=3-@XAlHyGMyq-$DiH z-D2pjxM5D*u;{N}5_o?H|Ip=hJ9BN2N)G3ht~a_C#Q^L9h=)c031rq>bFowLm>P=I|FS}^$q#wOVEErsGwifP4M=%_u;qOD@!dq~$7Tk5? zVtS)F2F_Q0u-^PtxLXX;J?UA?NoSr8SlycAl_%G3mmN$$s7+f^7D*Crm~=4Z)>;RW z@-3wsA4<4?XPj8<(j0Rm-e~Cs$QEoG7m7ykWHuHj`TxFmI|n6tT=<4CX4CJZycv;P z^&FO!c-4!TT+ zYSJ+&OxjHWW28ouo2e0|non9M!V}DwMIlCA%VlaUQ*e9Grs{Et94^K_bRSnGqN2%y z-fyKyyiHtW6Nc$25|HIV=i;Hm*ZfoRblddUY~$Isrtpo%$8--^jz;XJ;$tfQiV7T( zVI35VeBg=P;rmEPL@M#RT>FyXaD?xd6Iglcgy*4;L*!@se5fhc`QP`nFThK+eWqsi z9KIs>S;LZ@a|DpC=&zhg&--d~9rr_rFGXVZW6 z`bV$N4&_687ew+Y`1}v;oI9lV=j*rUL(eXV&mqM5#lzD)Rz^viVVa;3(d*JF|AM$4 zi@old^j!1Z4~BkAS(@SBKCI_L>nQl#uvoJRIU5K*H#|}!*!ln9ZQsn@vH3fyxxe1* z?x?W;b?f?$u>GfD2kJkq_I5PdfBFQEQWLw8VH;0&cf)|Jp;IPRS$PMACUik|qqzw` z9(iDJN?D1TSTBP^Cq1n&=2409bwKMu{o??9acYDMmH$kTyHh12e6b$2VIH7K{K_=t*)RJ2f0g#rm-WAxE7bD1b)ttj84;$13+sK`)JOT`u{wo^gU zNoT!Z&TFp|9El*ta2YRWe2|EY{HL_i(_=_yAL_toT6zi`sWyynJ`(pL6MiHS+pm0v zHgHW&9YzVN=K0Ts4ZQG=U*kD`)8BEyUvl1Ga_(PpfnRc-|Het+Cp8VTm*+w|bNb++DxQ~?MIT?fyxz-)mTPVN=4DSAe}rFdEyM5DFp4HWA6}N6 zsFis5qx@3MbNuE-PwkQ&_XnZUB?sr2lof=vLy%Z z#$c-HgQ=zurWyv9E&EYhZs)ytx!R4q|KSn-Dc-+?#5KV;k@?JCea};N export): + alias -> alias + link_type (LINK/CUSTOM) -> link_type ("redirect" / "custom") + original_url -> target_url + text -> content_markdown + description -> description + tags (M2M -> Tag) -> tags (list of tag names) + created_at -> created_at + updated_at -> updated_at + +Owner handling: + The current Django Link model has no owner/user FK. To stay forward- + compatible with older dumps that did carry an owner column, the script + introspects the `links_link` table for an owner-like FK + (`owner_id` / `user_id` / `created_by_id`) and, when present, joins it to + `auth_user` so `--owner-email` can filter. When no such column exists, + `--owner-email` prints a warning to stderr and exports all links. +""" +from __future__ import annotations + +import argparse +import json +import os +import sqlite3 +import sys +from typing import Any, Iterable + +# Django LinkType choices -> export vocabulary. +LINK_TYPE_MAP = {"LINK": "redirect", "CUSTOM": "custom"} +DEFAULT_LINK_TYPE = "redirect" + +# Columns on `links_link` that, if present, are treated as an owner FK into +# auth_user.id (checked in this order; first match wins). +OWNER_COLUMNS = ("owner_id", "user_id", "created_by_id") + +# Columns we expect to read from the link table. Names match the Django +# model; we tolerate missing columns by falling back to None. +LINK_COLUMNS = ( + "id", + "alias", + "link_type", + "original_url", + "text", + "description", + "created_at", + "updated_at", +) + + +def _stderr(msg: str) -> None: + print(msg, file=sys.stderr) + + +def resolve_db_path(arg_db: str | None) -> str: + """Resolve the DB path from --db, then $DJANGO_DB_PATH. Exit on missing.""" + db_path = arg_db or os.environ.get("DJANGO_DB_PATH") + if not db_path: + _stderr("error: no DB path provided. Use --db or set DJANGO_DB_PATH.") + sys.exit(2) + if not os.path.exists(db_path): + _stderr(f"error: DB file not found: {db_path}") + sys.exit(2) + if os.path.isdir(db_path): + _stderr(f"error: DB path is a directory, not a file: {db_path}") + sys.exit(2) + return db_path + + +def open_db(db_path: str) -> sqlite3.Connection: + """Open the SQLite DB read-only so we can never mutate the old DB.""" + uri = f"file:{db_path}?mode=ro" + try: + conn = sqlite3.connect(uri, uri=True) + except sqlite3.OperationalError: + # Very old / locked DBs may reject read-only URI; fall back to a + # plain read-only connection via immutability, never writing. + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA query_only = 1") # belt-and-suspenders: no writes + return conn + + +def table_columns(conn: sqlite3.Connection, table: str) -> list[str]: + rows = conn.execute(f'PRAGMA table_info("{table}")').fetchall() + return [r["name"] for r in rows] + + +def discover_link_table(conn: sqlite3.Connection) -> str: + """Find the links_link table via django_content_type, else a schema scan.""" + try: + ct = conn.execute( + "SELECT app_label, model FROM django_content_type " + "WHERE app_label = ? AND model = ?", + ("links", "link"), + ).fetchone() + except sqlite3.OperationalError: + ct = None + if ct: + return f"{ct['app_label']}_{ct['model']}" # => "links_link" + # Fallback: any table containing a unique-ish "alias" column. + tables = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ).fetchall() + for t in tables: + cols = table_columns(conn, t["name"]) + if "alias" in cols and "original_url" in cols: + return t["name"] + raise LookupError("could not find the links_link table in this DB") + + +def discover_tag_table(conn: sqlite3.Connection) -> str | None: + """Find the links_tag table (tag names), if it exists.""" + try: + ct = conn.execute( + "SELECT app_label, model FROM django_content_type " + "WHERE app_label = ? AND model = ?", + ("links", "tag"), + ).fetchone() + except sqlite3.OperationalError: + ct = None + if ct: + return f"{ct['app_label']}_{ct['model']}".replace(" ", "_") + # Fallback: a table with a `name` column in the links app namespace. + tables = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'links_%'" + ).fetchall() + for t in tables: + cols = table_columns(conn, t["name"]) + if "name" in cols and "id" in cols and "alias" not in cols: + return t["name"] + return None + + +def discover_m2m_table(conn: sqlite3.Connection, link_table: str, tag_table: str) -> str | None: + """Find the links_link -> links_tag M2M through table, if any.""" + # Django default name: links_link_tags + candidates = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' " + "AND (name LIKE ? OR name LIKE ?)", + (f"{link_table}_%", "%_tags"), + ).fetchall() + for t in candidates: + name = t["name"] + cols = table_columns(conn, name) + # An M2M through table has FKs to both sides. Django names them + # (or link) and (or tag). + has_link_fk = any(c in cols for c in ("link_id", link_table + "_id")) + has_tag_fk = any(c in cols for c in ("tag_id", tag_table + "_id")) + if has_link_fk and has_tag_fk: + return name + return None + + +def load_users(conn: sqlite3.Connection) -> dict[int, str]: + """Map auth_user.id -> email (lowercased) for owner filtering.""" + users: dict[int, str] = {} + try: + rows = conn.execute("SELECT id, email FROM auth_user").fetchall() + except sqlite3.OperationalError: + try: + rows = conn.execute("SELECT id, username FROM auth_user").fetchall() + # treat username as the "email"-ish identifier when email missing + except sqlite3.OperationalError: + return users + for r in rows: + users[r["id"]] = (r["email"] if "email" in r.keys() else r["username"]).lower() + return users + + +def detect_owner_column(conn: sqlite3.Connection, link_table: str) -> str | None: + """Return the owner FK column name on the link table, if any.""" + cols = table_columns(conn, link_table) + for c in OWNER_COLUMNS: + if c in cols: + return c + return None + + +def link_table_has_link_id_column(conn: sqlite3.Connection, m2m_table: str, link_table: str) -> str: + """Pick the FK column on the M2M table that points at the link row.""" + cols = table_columns(conn, m2m_table) + for cand in (link_table + "_id", "link_id"): + if cand in cols: + return cand + # last resort: the first _id column + for c in cols: + if c.endswith("_id") and c != "tag_id" and not c.endswith("tag_id"): + return c + raise LookupError(f"could not find link FK on M2M table {m2m_table}") + + +def query_links( + conn: sqlite3.Connection, + link_table: str, + tag_table: str | None, + m2m_table: str | None, + owner_col: str | None, + owner_email: str | None, + users: dict[int, str], +) -> Iterable[sqlite3.Row]: + """Build and run the SELECT, applying the owner filter when possible.""" + if owner_email and owner_col is None: + _stderr( + "warning: --owner-email given but the links_link table has no " + "owner/user FK column; owner filter cannot be applied. " + "Exporting all links." + ) + + # Assemble column list defensively (some columns may not exist in older + # DBs); we read only the ones present. + cols = table_columns(conn, link_table) + select_cols = [c for c in LINK_COLUMNS if c in cols] + if owner_col and owner_col in cols and owner_col not in select_cols: + select_cols.append(owner_col) + + base = f'SELECT {", ".join(select_cols)} FROM "{link_table}"' + where_parts: list[str] = [] + params: list[Any] = [] + + # Validate owner email against known users when an owner column exists. + if owner_email and owner_col is not None: + owner_email_lc = owner_email.strip().lower() + if not users: + _stderr( + "warning: --owner-email given but auth_user table is empty " + "or missing; no links will match the owner filter." + ) + where_parts.append("1=0") + else: + owner_user_ids = [uid for uid, em in users.items() if em == owner_email_lc] + if not owner_user_ids: + _stderr( + f"warning: no auth_user row matches --owner-email " + f"{owner_email!r}; export will be empty." + ) + where_parts.append("1=0") + else: + placeholders = ",".join("?" for _ in owner_user_ids) + where_parts.append(f"{owner_col} IN ({placeholders})") + params.extend(owner_user_ids) + + sql = base + if where_parts: + sql += " WHERE " + " AND ".join(where_parts) + sql += " ORDER BY alias ASC" + return conn.execute(sql, params) + + +def collect_tags( + conn: sqlite3.Connection, + link_id: int, + tag_table: str | None, + m2m_table: str | None, + link_fk_col: str, +) -> list[str]: + if not tag_table or not m2m_table: + return [] + rows = conn.execute( + f'SELECT t."name" FROM "{tag_table}" t ' + f'JOIN "{m2m_table}" m ON m."tag_id" = t."id" ' + f'WHERE m."{link_fk_col}" = ? ORDER BY t."name" ASC', + (link_id,), + ).fetchall() + return [r["name"] for r in rows] + + +def row_to_record( + conn: sqlite3.Connection, + row: sqlite3.Row, + tag_table: str | None, + m2m_table: str | None, + link_fk_col: str, +) -> dict[str, Any]: + link_id = row["id"] if "id" in row.keys() else None + + def g(col: str) -> Any: + return row[col] if col in row.keys() else None + + raw_link_type = g("link_type") + if raw_link_type in LINK_TYPE_MAP: + link_type = LINK_TYPE_MAP[raw_link_type] + elif not raw_link_type: + link_type = DEFAULT_LINK_TYPE + else: + link_type = raw_link_type + + tags = collect_tags(conn, link_id, tag_table, m2m_table, link_fk_col) if link_id is not None else [] + + return { + "alias": g("alias"), + "link_type": link_type, + "target_url": g("original_url"), + "content_markdown": g("text"), + "description": g("description"), + "tags": tags, + "created_at": g("created_at"), + "updated_at": g("updated_at"), + } + + +def emit(records: Iterable[dict[str, Any]], out_path: str | None) -> int: + """Write records as JSONL to out_path or stdout. Returns count written.""" + fh: Any = sys.stdout + if out_path and out_path != "-": + fh = open(out_path, "w", encoding="utf-8") + count = 0 + for rec in records: + fh.write(json.dumps(rec, ensure_ascii=False, default=str) + "\n") + count += 1 + if fh is not sys.stdout: + fh.close() + return count + + +def build_arg_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + prog="export-django-links", + description=( + "Export shortlinks from an old Django 'links' app SQLite DB to " + "JSONL. Standalone; does not modify the Django project." + ), + ) + p.add_argument( + "--db", + metavar="PATH", + default=None, + help="Path to the Django SQLite DB file (overrides $DJANGO_DB_PATH).", + ) + p.add_argument( + "--output", + "-o", + metavar="PATH", + default=None, + help=( + "Output JSONL file path. Use '-' or omit for stdout (default)." + ), + ) + p.add_argument( + "--owner-email", + metavar="EMAIL", + default=None, + help=( + "Only export links owned by this admin user (by auth_user email). " + "Requires an owner/user FK column on the links table; if none " + "exists, warns and exports all links." + ), + ) + return p + + +def main(argv: list[str] | None = None) -> int: + args = build_arg_parser().parse_args(argv) + db_path = resolve_db_path(args.db) + conn = open_db(db_path) + + try: + link_table = discover_link_table(conn) + except LookupError as e: + _stderr(f"error: {e}") + conn.close() + return 1 + + tag_table = discover_tag_table(conn) + m2m_table = ( + discover_m2m_table(conn, link_table, tag_table) if tag_table else None + ) + + users = load_users(conn) + owner_col = detect_owner_column(conn, link_table) + + link_fk_col = "link_id" + if m2m_table: + try: + link_fk_col = link_table_has_link_id_column(conn, m2m_table, link_table) + except LookupError: + link_fk_col = "link_id" + + rows = query_links( + conn, + link_table, + tag_table, + m2m_table, + owner_col, + args.owner_email, + users, + ) + + records = ( + row_to_record(conn, r, tag_table, m2m_table, link_fk_col) for r in rows + ) + count = emit(records, args.output) + + conn.close() + + if args.output and args.output != "-": + _stderr(f"exported {count} link(s) to {args.output}") + else: + _stderr(f"exported {count} link(s) to stdout") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/export-django-links.test.py b/scripts/export-django-links.test.py new file mode 100644 index 0000000..b0819f5 --- /dev/null +++ b/scripts/export-django-links.test.py @@ -0,0 +1,340 @@ +#!/usr/bin/env python3 +""" +Tests for scripts/export-django-links.py. + +Run: + python3 scripts/export-django-links.test.py +or + pytest scripts/export-django-links.test.py + +These tests build a throwaway SQLite DB that mimics the old Django `links` +app schema (auth_user, django_content_type, links_link, links_tag, +links_link_tags) and asserts the exporter reads it correctly. +""" +from __future__ import annotations + +import io +import json +import os +import sqlite3 +import tempfile +import unittest +from contextlib import redirect_stderr, redirect_stdout + +# Make the script importable by path (scripts/ is not a package and the +# filename contains hyphens, so a normal `import` won't work). +HERE = os.path.dirname(os.path.abspath(__file__)) +_SCRIPT_PATH = os.path.join(HERE, "export-django-links.py") + +import importlib.util # noqa: E402 + +_spec = importlib.util.spec_from_file_location("export_django_links", _SCRIPT_PATH) +assert _spec is not None and _spec.loader is not None +exporter_mod = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(exporter_mod) + + +def _make_fixture(db_path: str, with_owner: bool = False) -> sqlite3.Connection: + """Create a Django-shaped links DB at db_path with a couple of rows.""" + conn = sqlite3.connect(db_path) + c = conn.cursor() + + c.execute("CREATE TABLE django_content_type (id INTEGER, app_label TEXT, model TEXT)") + c.execute("INSERT INTO django_content_type (id, app_label, model) VALUES (1,'links','link')") + c.execute("INSERT INTO django_content_type (id, app_label, model) VALUES (2,'links','tag')") + + c.execute( + "CREATE TABLE auth_user (id INTEGER PRIMARY KEY, username TEXT, email TEXT, is_staff INTEGER)" + ) + c.execute("INSERT INTO auth_user (id, username, email, is_staff) VALUES (1,'alice','alice@example.com',1)") + c.execute("INSERT INTO auth_user (id, username, email, is_staff) VALUES (2,'bob','bob@example.com',1)") + + link_cols = [ + "id INTEGER PRIMARY KEY", + "alias TEXT UNIQUE", + "original_url TEXT", + "text TEXT", + "link_type TEXT", + "click_count INTEGER DEFAULT 0", + "description TEXT", + "created_at TEXT", + "updated_at TEXT", + ] + if with_owner: + link_cols.append("owner_id INTEGER REFERENCES auth_user(id)") + c.execute(f"CREATE TABLE links_link ({', '.join(link_cols)})") + + c.execute( + "CREATE TABLE links_tag (id INTEGER PRIMARY KEY, name TEXT UNIQUE, slug TEXT, " + "description TEXT, created_at TEXT, updated_at TEXT)" + ) + + c.execute( + "CREATE TABLE links_link_tags (id INTEGER PRIMARY KEY, link_id INTEGER, tag_id INTEGER)" + ) + + # Link 1: redirect with two tags, owned by alice (when with_owner). + c.execute( + "INSERT INTO links_link (id, alias, original_url, text, link_type, description, " + "created_at, updated_at, owner_id) " + "VALUES (1,'go','https://example.com/go','','LINK','A redirect'," + "'2024-01-01 00:00:00','2024-01-02 00:00:00',1)" + if with_owner else + "INSERT INTO links_link (id, alias, original_url, text, link_type, description, " + "created_at, updated_at) " + "VALUES (1,'go','https://example.com/go','','LINK','A redirect'," + "'2024-01-01 00:00:00','2024-01-02 00:00:00')" + ) + c.execute("INSERT INTO links_tag (id, name, slug) VALUES (1,'work','work')") + c.execute("INSERT INTO links_tag (id, name, slug) VALUES (2,'search','search')") + c.execute("INSERT INTO links_link_tags (id, link_id, tag_id) VALUES (1,1,1)") + c.execute("INSERT INTO links_link_tags (id, link_id, tag_id) VALUES (2,1,2)") + + # Link 2: custom (markdown content), owned by bob (when with_owner). + c.execute( + "INSERT INTO links_link (id, alias, original_url, text, link_type, description, " + "created_at, updated_at, owner_id) " + "VALUES (2,'note','','# Notes','CUSTOM','A custom page'," + "'2024-02-01 00:00:00','2024-02-03 00:00:00',2)" + if with_owner else + "INSERT INTO links_link (id, alias, original_url, text, link_type, description, " + "created_at, updated_at) " + "VALUES (2,'note','','# Notes','CUSTOM','A custom page'," + "'2024-02-01 00:00:00','2024-02-03 00:00:00')" + ) + c.execute("INSERT INTO links_tag (id, name, slug) VALUES (3,'personal','personal')") + c.execute("INSERT INTO links_link_tags (id, link_id, tag_id) VALUES (3,2,3)") + + conn.commit() + return conn + + +def _read_jsonl(path: str) -> list[dict]: + out = [] + with open(path, "r", encoding="utf-8") as fh: + for line in fh: + line = line.rstrip("\n") + if line: + out.append(json.loads(line)) + return out + + +class ExportTests(unittest.TestCase): + def setUp(self): + self.tmpdir = tempfile.mkdtemp(prefix="heygo-export-test-") + self.db_path = os.path.join(self.tmpdir, "db.sqlite3") + + def tearDown(self): + # Clear env so it doesn't leak between tests. + os.environ.pop("DJANGO_DB_PATH", None) + + def _run(self, argv: list[str]) -> tuple[int, str, str]: + out = io.StringIO() + err = io.StringIO() + try: + with redirect_stdout(out), redirect_stderr(err): + code = exporter_mod.main(argv) + return code, out.getvalue(), err.getvalue() + except SystemExit as e: + code = e.code if e.code is not None else 1 + return int(code), out.getvalue(), err.getvalue() + + def test_basic_export_to_file(self): + _make_fixture(self.db_path, with_owner=False) + out_path = os.path.join(self.tmpdir, "out.jsonl") + code, _, err = self._run(["--db", self.db_path, "--output", out_path]) + self.assertEqual(code, 0) + rows = _read_jsonl(out_path) + self.assertEqual(len(rows), 2) + # Ordered by alias ASC: 'go' then 'note' + self.assertEqual(rows[0]["alias"], "go") + self.assertEqual(rows[1]["alias"], "note") + # Field mapping for the redirect link + self.assertEqual(rows[0]["link_type"], "redirect") + self.assertEqual(rows[0]["target_url"], "https://example.com/go") + self.assertEqual(rows[0]["content_markdown"], "") + self.assertEqual(rows[0]["description"], "A redirect") + self.assertEqual(rows[0]["tags"], ["search", "work"]) # sorted asc by name + self.assertEqual(rows[0]["created_at"], "2024-01-01 00:00:00") + self.assertEqual(rows[0]["updated_at"], "2024-01-02 00:00:00") + # Field mapping for the custom link + self.assertEqual(rows[1]["link_type"], "custom") + self.assertEqual(rows[1]["content_markdown"], "# Notes") + self.assertEqual(rows[1]["tags"], ["personal"]) + self.assertIn("exported 2 link(s)", err) + + def test_link_type_mapping_defaults(self): + # link_type null -> redirect; unknown value preserved as-is + conn = _make_fixture(self.db_path, with_owner=False) + conn.execute( + "INSERT INTO links_link (id, alias, original_url, text, link_type, created_at, updated_at) " + "VALUES (3,'bare','https://example.com/bare',NULL,NULL,'2024-03-01','2024-03-01')" + ) + conn.execute( + "INSERT INTO links_link (id, alias, original_url, text, link_type, created_at, updated_at) " + "VALUES (4,'weird','https://example.com/weird','','BOGUS','2024-03-02','2024-03-02')" + ) + conn.commit() + conn.close() + out_path = os.path.join(self.tmpdir, "out.jsonl") + code, _, _ = self._run(["--db", self.db_path, "--output", out_path]) + self.assertEqual(code, 0) + rows = _read_jsonl(out_path) + by_alias = {r["alias"]: r for r in rows} + self.assertEqual(by_alias["bare"]["link_type"], "redirect") + self.assertEqual(by_alias["weird"]["link_type"], "BOGUS") + + def test_owner_filter_with_owner_column(self): + _make_fixture(self.db_path, with_owner=True) + out_path = os.path.join(self.tmpdir, "out.jsonl") + code, _, err = self._run( + ["--db", self.db_path, "--output", out_path, "--owner-email", "alice@example.com"] + ) + self.assertEqual(code, 0) + rows = _read_jsonl(out_path) + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["alias"], "go") # owned by alice (id=1) + self.assertIn("exported 1 link(s)", err) + + def test_owner_filter_email_case_insensitive(self): + _make_fixture(self.db_path, with_owner=True) + out_path = os.path.join(self.tmpdir, "out.jsonl") + code, _, _ = self._run( + ["--db", self.db_path, "--output", out_path, "--owner-email", "ALICE@example.COM"] + ) + self.assertEqual(code, 0) + rows = _read_jsonl(out_path) + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["alias"], "go") + + def test_owner_filter_no_match_is_empty(self): + _make_fixture(self.db_path, with_owner=True) + out_path = os.path.join(self.tmpdir, "out.jsonl") + code, _, err = self._run( + ["--db", self.db_path, "--output", out_path, "--owner-email", "nobody@example.com"] + ) + self.assertEqual(code, 0) + rows = _read_jsonl(out_path) + self.assertEqual(rows, []) + self.assertIn("no auth_user row matches", err) + + def test_owner_filter_without_owner_column_warns_and_exports_all(self): + _make_fixture(self.db_path, with_owner=False) + out_path = os.path.join(self.tmpdir, "out.jsonl") + code, _, err = self._run( + ["--db", self.db_path, "--output", out_path, "--owner-email", "alice@example.com"] + ) + self.assertEqual(code, 0) + rows = _read_jsonl(out_path) + self.assertEqual(len(rows), 2) # no owner column -> exported all + self.assertIn("owner filter cannot be applied", err) + + def test_env_var_db_path(self): + _make_fixture(self.db_path, with_owner=False) + os.environ["DJANGO_DB_PATH"] = self.db_path + out_path = os.path.join(self.tmpdir, "out.jsonl") + code, _, _ = self._run(["--output", out_path]) # no --db + self.assertEqual(code, 0) + rows = _read_jsonl(out_path) + self.assertEqual(len(rows), 2) + + def test_db_arg_overrides_env_var(self): + _make_fixture(self.db_path, with_owner=False) + other = os.path.join(self.tmpdir, "empty.sqlite3") + sqlite3.connect(other).close() # empty DB (no tables) -> error + os.environ["DJANGO_DB_PATH"] = other + out_path = os.path.join(self.tmpdir, "out.jsonl") + code, _, _ = self._run(["--db", self.db_path, "--output", out_path]) + self.assertEqual(code, 0) + rows = _read_jsonl(out_path) + self.assertEqual(len(rows), 2) + + def test_missing_db_path_exits_nonzero(self): + code, _, err = self._run(["--db", "/no/such/file.sqlite3"]) + self.assertEqual(code, 2) + self.assertIn("DB file not found", err) + + def test_no_db_path_arg_or_env_exits_nonzero(self): + os.environ.pop("DJANGO_DB_PATH", None) + code, _, err = self._run([]) + self.assertEqual(code, 2) + self.assertIn("no DB path provided", err) + + def test_db_path_is_directory_exits_nonzero(self): + code, _, err = self._run(["--db", self.tmpdir]) + self.assertEqual(code, 2) + self.assertIn("is a directory", err) + + def test_empty_links_table_exports_zero(self): + conn = _make_fixture(self.db_path, with_owner=False) + conn.execute("DELETE FROM links_link") + conn.execute("DELETE FROM links_link_tags") + conn.commit() + conn.close() + out_path = os.path.join(self.tmpdir, "out.jsonl") + code, _, err = self._run(["--db", self.db_path, "--output", out_path]) + self.assertEqual(code, 0) + rows = _read_jsonl(out_path) + self.assertEqual(rows, []) + self.assertIn("exported 0 link(s)", err) + + def test_stdout_default_output(self): + _make_fixture(self.db_path, with_owner=False) + code, out, _ = self._run(["--db", self.db_path]) + self.assertEqual(code, 0) + lines = [ln for ln in out.splitlines() if ln.strip()] + self.assertEqual(len(lines), 2) + first = json.loads(lines[0]) + self.assertEqual(first["alias"], "go") + + def test_no_tables_in_db_exits_nonzero(self): + empty = os.path.join(self.tmpdir, "empty.sqlite3") + sqlite3.connect(empty).close() + out_path = os.path.join(self.tmpdir, "out.jsonl") + code, _, err = self._run(["--db", empty, "--output", out_path]) + self.assertEqual(code, 1) + self.assertIn("could not find the links_link table", err) + + def test_table_discovery_fallback_without_content_type(self): + # Drop django_content_type to exercise the sqlite_master fallback. + conn = _make_fixture(self.db_path, with_owner=False) + conn.execute("DROP TABLE django_content_type") + conn.commit() + conn.close() + out_path = os.path.join(self.tmpdir, "out.jsonl") + code, _, _ = self._run(["--db", self.db_path, "--output", out_path]) + self.assertEqual(code, 0) + rows = _read_jsonl(out_path) + self.assertEqual(len(rows), 2) + + def test_tags_absent_when_no_tag_table(self): + conn = _make_fixture(self.db_path, with_owner=False) + conn.execute("DROP TABLE links_link_tags") + conn.execute("DROP TABLE links_tag") + conn.commit() + conn.close() + out_path = os.path.join(self.tmpdir, "out.jsonl") + code, _, _ = self._run(["--db", self.db_path, "--output", out_path]) + self.assertEqual(code, 0) + rows = _read_jsonl(out_path) + self.assertEqual(len(rows), 2) + for r in rows: + self.assertEqual(r["tags"], []) + + +class ArgParserTests(unittest.TestCase): + def test_help_lists_all_options(self): + p = exporter_mod.build_arg_parser() + h = p.format_help() + for token in ("--db", "--output", "--owner-email", "DJANGO_DB_PATH"): + self.assertIn(token, h) + + def test_defaults(self): + ns = exporter_mod.build_arg_parser().parse_args([]) + self.assertIsNone(ns.db) + self.assertIsNone(ns.output) + self.assertIsNone(ns.owner_email) + + +if __name__ == "__main__": + unittest.main(verbosity=2)