smscalls: make some fields optional, yield errors
reflects the new types-lxml package https://github.com/abelcheung/types-lxml
This commit is contained in:
parent
d464b1e607
commit
02c738594f
1 changed files with 33 additions and 17 deletions
|
@ -18,45 +18,54 @@ config = make_config(smscalls)
|
||||||
|
|
||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import NamedTuple, Iterator, Set, Tuple
|
from typing import NamedTuple, Iterator, Set, Tuple, Optional
|
||||||
|
|
||||||
from lxml import etree # type: ignore
|
from lxml import etree # type: ignore
|
||||||
|
|
||||||
from .core.common import get_files, Stats
|
from my.core.common import get_files, Stats
|
||||||
|
from my.core.error import Res
|
||||||
|
|
||||||
|
|
||||||
class Call(NamedTuple):
|
class Call(NamedTuple):
|
||||||
dt: datetime
|
dt: datetime
|
||||||
dt_readable: str
|
dt_readable: Optional[str]
|
||||||
duration_s: int
|
duration_s: Optional[int]
|
||||||
who: str
|
who: Optional[str]
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def summary(self) -> str:
|
def summary(self) -> str:
|
||||||
return f"talked with {self.who} for {self.duration_s} secs"
|
return f"talked with {self.who} for {self.duration_s} secs"
|
||||||
|
|
||||||
|
|
||||||
def _extract_calls(path: Path) -> Iterator[Call]:
|
def _extract_calls(path: Path) -> Iterator[Res[Call]]:
|
||||||
tr = etree.parse(str(path))
|
tr = etree.parse(str(path))
|
||||||
for cxml in tr.findall('call'):
|
for cxml in tr.findall('call'):
|
||||||
|
date_str = cxml.get('date')
|
||||||
|
if date_str is None:
|
||||||
|
yield RuntimeError(f"no date in {etree.tostring(cxml).decode('utf-8')}")
|
||||||
|
continue
|
||||||
|
duration = cxml.get('duration')
|
||||||
# TODO we've got local tz here, not sure if useful..
|
# TODO we've got local tz here, not sure if useful..
|
||||||
# ok, so readable date is local datetime, changing throughout the backup
|
# ok, so readable date is local datetime, changing throughout the backup
|
||||||
yield Call(
|
yield Call(
|
||||||
dt=_parse_dt_ms(cxml.get('date')),
|
dt=_parse_dt_ms(date_str),
|
||||||
dt_readable=cxml.get('readable_date'),
|
dt_readable=cxml.get('readable_date'),
|
||||||
duration_s=int(cxml.get('duration')),
|
duration_s=int(duration) if duration is not None else None,
|
||||||
who=cxml.get('contact_name') # TODO number if contact is unavail??
|
who=cxml.get('contact_name') # TODO number if contact is unavail??
|
||||||
# TODO type? must be missing/outgoing/incoming
|
# TODO type? must be missing/outgoing/incoming
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def calls() -> Iterator[Call]:
|
def calls() -> Iterator[Res[Call]]:
|
||||||
files = get_files(config.export_path, glob='calls-*.xml')
|
files = get_files(config.export_path, glob='calls-*.xml')
|
||||||
|
|
||||||
# TODO always replacing with the latter is good, we get better contact names??
|
# TODO always replacing with the latter is good, we get better contact names??
|
||||||
emitted: Set[datetime] = set()
|
emitted: Set[datetime] = set()
|
||||||
for p in files:
|
for p in files:
|
||||||
for c in _extract_calls(p):
|
for c in _extract_calls(p):
|
||||||
|
if isinstance(c, Exception):
|
||||||
|
yield c
|
||||||
|
continue
|
||||||
if c.dt in emitted:
|
if c.dt in emitted:
|
||||||
continue
|
continue
|
||||||
emitted.add(c.dt)
|
emitted.add(c.dt)
|
||||||
|
@ -65,19 +74,22 @@ def calls() -> Iterator[Call]:
|
||||||
|
|
||||||
class Message(NamedTuple):
|
class Message(NamedTuple):
|
||||||
dt: datetime
|
dt: datetime
|
||||||
dt_readable: str
|
dt_readable: Optional[str]
|
||||||
who: str
|
who: Optional[str]
|
||||||
message: str
|
message: Optional[str]
|
||||||
phone_number: str
|
phone_number: Optional[str]
|
||||||
from_me: bool
|
from_me: bool
|
||||||
|
|
||||||
|
|
||||||
def messages() -> Iterator[Message]:
|
def messages() -> Iterator[Res[Message]]:
|
||||||
files = get_files(config.export_path, glob='sms-*.xml')
|
files = get_files(config.export_path, glob='sms-*.xml')
|
||||||
|
|
||||||
emitted: Set[Tuple[datetime, str, bool]] = set()
|
emitted: Set[Tuple[datetime, Optional[str], Optional[bool]]] = set()
|
||||||
for p in files:
|
for p in files:
|
||||||
for c in _extract_messages(p):
|
for c in _extract_messages(p):
|
||||||
|
if isinstance(c, Exception):
|
||||||
|
yield c
|
||||||
|
continue
|
||||||
key = (c.dt, c.who, c.from_me)
|
key = (c.dt, c.who, c.from_me)
|
||||||
if key in emitted:
|
if key in emitted:
|
||||||
continue
|
continue
|
||||||
|
@ -85,11 +97,15 @@ def messages() -> Iterator[Message]:
|
||||||
yield c
|
yield c
|
||||||
|
|
||||||
|
|
||||||
def _extract_messages(path: Path) -> Iterator[Message]:
|
def _extract_messages(path: Path) -> Iterator[Res[Message]]:
|
||||||
tr = etree.parse(str(path))
|
tr = etree.parse(str(path))
|
||||||
for mxml in tr.findall('sms'):
|
for mxml in tr.findall('sms'):
|
||||||
|
date_str = mxml.get('date')
|
||||||
|
if date_str is None:
|
||||||
|
yield RuntimeError(f"no date in {etree.tostring(mxml).decode('utf-8')}")
|
||||||
|
continue
|
||||||
yield Message(
|
yield Message(
|
||||||
dt=_parse_dt_ms(mxml.get('date')),
|
dt=_parse_dt_ms(date_str),
|
||||||
dt_readable=mxml.get('readable_date'),
|
dt_readable=mxml.get('readable_date'),
|
||||||
who=mxml.get('contact_name'),
|
who=mxml.get('contact_name'),
|
||||||
message=mxml.get('body'),
|
message=mxml.get('body'),
|
||||||
|
|
Loading…
Add table
Reference in a new issue