52 lines
2.3 KiB
Python
52 lines
2.3 KiB
Python
|
"""Export all locally stored contact records through WhatsApp's JavaScript API."""
|
||
|
|
import argparse
|
||
|
|
import datetime as dt
|
||
|
|
import json
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from .live import run_math
|
||
|
|
from .zip_export import evaluate
|
||
|
|
|
||
|
|
PROBE = r'''(async () => {
|
||
|
|
const table=require('WAWebSchemaContact_DO_NOT_USE_DIRECTLY').getContactTable();
|
||
|
|
const rows=await table.all();
|
||
|
|
if(!Array.isArray(rows))throw Error('Unexpected contact table result');
|
||
|
|
const fields=['id','phoneNumber','name','shortName','pushname','notifyName','verifiedName',
|
||
|
|
'addressBookName','isAddressBookContact','isMyContact','isBusiness','isEnterprise'];
|
||
|
|
const contacts=rows.map(row=>{
|
||
|
|
const out={};for(const field of fields){const value=row[field];
|
||
|
|
if(value==null||['string','number','boolean'].includes(typeof value))out[field]=value??null;
|
||
|
|
else if(field==='id')out.id=value.toString();
|
||
|
|
}
|
||
|
|
if(typeof out.id!=='string'||!out.id)throw Error('Contact ID unavailable');
|
||
|
|
return out;
|
||
|
|
});
|
||
|
|
return {source:'WAWebSchemaContact_DO_NOT_USE_DIRECTLY.getContactTable().all',
|
||
|
|
scope:'All local contact rows, not only loaded models or phone address-book entries',
|
||
|
|
count:contacts.length,contacts,cdp:false,ui_automation:false};
|
||
|
|
})()'''
|
||
|
|
|
||
|
|
|
||
|
|
def export_contacts(output,pid=None):
|
||
|
|
output=Path(output)
|
||
|
|
if output.exists():raise ValueError('Output exists; choose a new --output')
|
||
|
|
def operation(execute,session,report):
|
||
|
|
data=evaluate(execute,PROBE,session)
|
||
|
|
if not isinstance(data.get('contacts'),list) or data.get('count')!=len(data['contacts']):
|
||
|
|
raise RuntimeError('Incomplete contact export response')
|
||
|
|
report['data']=data
|
||
|
|
report=run_math(pid,'contacts-export',_operation=operation)
|
||
|
|
output.parent.mkdir(parents=True,exist_ok=True)
|
||
|
|
with output.open('x',encoding='utf-8') as stream:
|
||
|
|
json.dump(report,stream,ensure_ascii=True,indent=2)
|
||
|
|
return report
|
||
|
|
|
||
|
|
|
||
|
|
def main(argv=None):
|
||
|
|
parser=argparse.ArgumentParser(description=__doc__)
|
||
|
|
parser.add_argument('--pid',type=int)
|
||
|
|
parser.add_argument('--output',type=Path,default=Path('exports')/('contacts-'+dt.datetime.now().strftime('%Y%m%d-%H%M%S')+'.json'))
|
||
|
|
args=parser.parse_args(argv)
|
||
|
|
report=export_contacts(args.output,args.pid)
|
||
|
|
print(json.dumps({'passed':report['passed'],'error':report.get('error'),'contacts':report.get('data',{}).get('count'),'output':str(args.output)}))
|
||
|
|
return 0 if report['passed'] else 1
|