relikd/abcddb2vcard
1
0
Fork 0
Code Pull Requests Releases Activity

fix: ignore non-existent columns

Browse Source
This commit is contained in:
relikd
2024-01-27 16:16:47 +01:00
parent 52cad1ebf9
commit f3ae2e3ff4
2 changed files with 37 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
README.md
View File

@@ -8,15 +8,21 @@ The output of this script should be exactly the same as dragging and dropping th
### Usage ### Usage
``` ```sh
python3 abcddb2vcard.py backup/contacts_$(date +"%Y-%m-%d").vcf python3 abcddb2vcard.py backup/contacts_$(date +"%Y-%m-%d").vcf
``` ```
> assuming db is located at "~/Library/Application Support/AddressBook/AddressBook-v22.abcddb" > assuming db is located at "~/Library/Application Support/AddressBook/AddressBook-v22.abcddb"
#### Export into individual files
```sh
python3 abcddb2vcard.py outdir -s 'path/%{fullname}.vcf'
```
#### Extract contact images #### Extract contact images
``` ```sh
python3 vcard2image.py AllContacts.vcf ./profile_pics/ python3 vcard2image.py AllContacts.vcf ./profile_pics/
``` ```

46
abcddb2vcard/ABCDDB.py
View File

@@ -6,7 +6,8 @@ from urllib.parse import quote
from typing import List, Dict, Any, Iterable, Optional from typing import List, Dict, Any, Iterable, Optional
ITEM_COUNTER = 0 ITEM_COUNTER = 0
rx_query = re.compile(r'SELECT([\s\S]*)FROM[\s]+([A-Z_]+)')
rx_cols = re.compile(r'[\s,;](Z[A-Z_]+)')
# =============================== # ===============================
# Helper methods # Helper methods
@@ -47,6 +48,17 @@ def buildLabel(
return incrItem(value, label) return incrItem(value, label)
def sanitize(cursor: sqlite3.Cursor, query: str) -> str:
cols, table = rx_query.findall(query)[0]
sel_cols = {x for x in rx_cols.findall(cols)}
all_cols = {x[1] for x in cursor.execute(f'PRAGMA table_info({table});')}
missing_cols = sel_cols.difference(all_cols)
for missing in missing_cols:
print(f'WARN: column "{missing}" not found in {table}. Ignoring.',
file=sys.stderr)
query = query.replace(missing, 'NULL')
return query
# =============================== # ===============================
# VCARD Attributes # VCARD Attributes
# =============================== # ===============================
@@ -74,10 +86,10 @@ class Queryable: # Protocol
class Email(Queryable): class Email(Queryable):
@staticmethod @staticmethod
def queryAll(cursor: sqlite3.Cursor) -> Iterable['Email']: def queryAll(cursor: sqlite3.Cursor) -> Iterable['Email']:
return (Email(x) for x in cursor.execute(''' return (Email(x) for x in cursor.execute(sanitize(cursor, '''
SELECT ZOWNER, ZLABEL, ZADDRESS SELECT ZOWNER, ZLABEL, ZADDRESS
FROM ZABCDEMAILADDRESS FROM ZABCDEMAILADDRESS
ORDER BY ZOWNER, ZISPRIMARY DESC, ZORDERINGINDEX;''')) ORDER BY ZOWNER, ZISPRIMARY DESC, ZORDERINGINDEX;''')))
def __init__(self, row: List[Any]): def __init__(self, row: List[Any]):
self._parent = row[0] # type: int self._parent = row[0] # type: int
@@ -95,10 +107,10 @@ class Email(Queryable):
class Phone(Queryable): class Phone(Queryable):
@staticmethod @staticmethod
def queryAll(cursor: sqlite3.Cursor) -> Iterable['Phone']: def queryAll(cursor: sqlite3.Cursor) -> Iterable['Phone']:
return (Phone(x) for x in cursor.execute(''' return (Phone(x) for x in cursor.execute(sanitize(cursor, '''
SELECT ZOWNER, ZLABEL, ZFULLNUMBER SELECT ZOWNER, ZLABEL, ZFULLNUMBER
FROM ZABCDPHONENUMBER FROM ZABCDPHONENUMBER
ORDER BY ZOWNER, ZISPRIMARY DESC, ZORDERINGINDEX;''')) ORDER BY ZOWNER, ZISPRIMARY DESC, ZORDERINGINDEX;''')))
def __init__(self, row: List[Any]): def __init__(self, row: List[Any]):
self._parent = row[0] # type: int self._parent = row[0] # type: int
@@ -131,11 +143,11 @@ class Phone(Queryable):
class Address(Queryable): class Address(Queryable):
@staticmethod @staticmethod
def queryAll(cursor: sqlite3.Cursor) -> Iterable['Address']: def queryAll(cursor: sqlite3.Cursor) -> Iterable['Address']:
return (Address(x) for x in cursor.execute(''' return (Address(x) for x in cursor.execute(sanitize(cursor, '''
SELECT ZOWNER, ZLABEL, SELECT ZOWNER, ZLABEL,
ZSTREET, ZCITY, ZSTATE, ZZIPCODE, ZCOUNTRYNAME ZSTREET, ZCITY, ZSTATE, ZZIPCODE, ZCOUNTRYNAME
FROM ZABCDPOSTALADDRESS FROM ZABCDPOSTALADDRESS
ORDER BY ZOWNER, ZISPRIMARY DESC, ZORDERINGINDEX;''')) ORDER BY ZOWNER, ZISPRIMARY DESC, ZORDERINGINDEX;''')))
def __init__(self, row: List[Any]): def __init__(self, row: List[Any]):
self._parent = row[0] # type: int self._parent = row[0] # type: int
@@ -160,9 +172,9 @@ class Address(Queryable):
class SocialProfile(Queryable): class SocialProfile(Queryable):
@staticmethod @staticmethod
def queryAll(cursor: sqlite3.Cursor) -> Iterable['SocialProfile']: def queryAll(cursor: sqlite3.Cursor) -> Iterable['SocialProfile']:
return (SocialProfile(x) for x in cursor.execute(''' return (SocialProfile(x) for x in cursor.execute(sanitize(cursor, '''
SELECT ZOWNER, ZSERVICENAME, ZUSERNAME SELECT ZOWNER, ZSERVICENAME, ZUSERNAME
FROM ZABCDSOCIALPROFILE;''')) FROM ZABCDSOCIALPROFILE;''')))
def __init__(self, row: List[Any]): def __init__(self, row: List[Any]):
self._parent = row[0] # type: int self._parent = row[0] # type: int
@@ -181,10 +193,10 @@ class SocialProfile(Queryable):
class Note(Queryable): class Note(Queryable):
@staticmethod @staticmethod
def queryAll(cursor: sqlite3.Cursor) -> Iterable['Note']: def queryAll(cursor: sqlite3.Cursor) -> Iterable['Note']:
return (Note(x) for x in cursor.execute(''' return (Note(x) for x in cursor.execute(sanitize(cursor, '''
SELECT ZCONTACT, ZTEXT SELECT ZCONTACT, ZTEXT
FROM ZABCDNOTE FROM ZABCDNOTE
WHERE ZTEXT IS NOT NULL;''')) WHERE ZTEXT IS NOT NULL;''')))
def __init__(self, row: List[Any]): def __init__(self, row: List[Any]):
self._parent = row[0] # type: int self._parent = row[0] # type: int
@@ -200,10 +212,10 @@ class Note(Queryable):
class URL(Queryable): class URL(Queryable):
@staticmethod @staticmethod
def queryAll(cursor: sqlite3.Cursor) -> Iterable['URL']: def queryAll(cursor: sqlite3.Cursor) -> Iterable['URL']:
return (URL(x) for x in cursor.execute(''' return (URL(x) for x in cursor.execute(sanitize(cursor, '''
SELECT ZOWNER, ZLABEL, ZURL SELECT ZOWNER, ZLABEL, ZURL
FROM ZABCDURLADDRESS FROM ZABCDURLADDRESS
ORDER BY ZOWNER, ZISPRIMARY DESC, ZORDERINGINDEX;''')) ORDER BY ZOWNER, ZISPRIMARY DESC, ZORDERINGINDEX;''')))
def __init__(self, row: List[Any]): def __init__(self, row: List[Any]):
self._parent = row[0] # type: int self._parent = row[0] # type: int
@@ -220,11 +232,11 @@ class URL(Queryable):
class Service(Queryable): class Service(Queryable):
@staticmethod @staticmethod
def queryAll(cursor: sqlite3.Cursor) -> Iterable['Service']: def queryAll(cursor: sqlite3.Cursor) -> Iterable['Service']:
return (Service(x) for x in cursor.execute(''' return (Service(x) for x in cursor.execute(sanitize(cursor, '''
SELECT ZOWNER, ZSERVICENAME, ZLABEL, ZADDRESS SELECT ZOWNER, ZSERVICENAME, ZLABEL, ZADDRESS
FROM ZABCDMESSAGINGADDRESS FROM ZABCDMESSAGINGADDRESS
INNER JOIN ZABCDSERVICE ON ZSERVICE = ZABCDSERVICE.Z_PK INNER JOIN ZABCDSERVICE ON ZSERVICE = ZABCDSERVICE.Z_PK
ORDER BY ZOWNER, ZISPRIMARY DESC, ZORDERINGINDEX;''')) ORDER BY ZOWNER, ZISPRIMARY DESC, ZORDERINGINDEX;''')))
def __init__(self, row: List[Any]): def __init__(self, row: List[Any]):
self._parent = row[0] # type: int self._parent = row[0] # type: int
@@ -281,7 +293,7 @@ class Record:
'SELECT Z_ENT FROM Z_PRIMARYKEY WHERE Z_NAME == "ABCDContact"' 'SELECT Z_ENT FROM Z_PRIMARYKEY WHERE Z_NAME == "ABCDContact"'
).fetchone()[0] ).fetchone()[0]
# find all records that match this id # find all records that match this id
return {x[0]: Record(x) for x in cursor.execute(''' return {x[0]: Record(x) for x in cursor.execute(sanitize(cursor, '''
SELECT Z_PK, SELECT Z_PK,
ZFIRSTNAME, ZLASTNAME, ZMIDDLENAME, ZTITLE, ZSUFFIX, ZFIRSTNAME, ZLASTNAME, ZMIDDLENAME, ZTITLE, ZSUFFIX,
ZNICKNAME, ZMAIDENNAME, ZNICKNAME, ZMAIDENNAME,
@@ -290,7 +302,7 @@ class Record:
strftime('%Y-%m-%d', ZBIRTHDAY + 978307200, 'unixepoch'), strftime('%Y-%m-%d', ZBIRTHDAY + 978307200, 'unixepoch'),
ZTHUMBNAILIMAGEDATA, ZDISPLAYFLAGS ZTHUMBNAILIMAGEDATA, ZDISPLAYFLAGS
FROM ZABCDRECORD FROM ZABCDRECORD
WHERE Z_ENT = ?;''', [z_ent])} WHERE Z_ENT = ?;'''), [z_ent])}
@staticmethod @staticmethod
def initEmpty(id: int) -> 'Record': def initEmpty(id: int) -> 'Record':