309 lines
12 KiB
Python
309 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
from typing import Self, Any, Optional, Sequence, Callable
|
|
from requests import get, Response
|
|
from parsel import Selector as ParseSelector, SelectorList as ParseSelectorList
|
|
from AnP.DSLs.ScrapDSL import (
|
|
Execute, Selector, CSS, XPath, Name, LinkText, Class, ID, Open, Close, Switch,
|
|
Go, Get, ForEach, Do, Condition, And, Or, Not, If, While, DoWhile, Download,
|
|
Action
|
|
)
|
|
from AnP.Interfaces.Application import AnPInterface
|
|
from AnP.Abstracts.BrowserAbstract import BrowserAbstract
|
|
from AnP.Models.XMLScrapTabModel import XMLScrapTabModel
|
|
|
|
class XMLScrapDriver(BrowserAbstract):
|
|
|
|
def __init__(self:Self, anp:AnPInterface, key:str, inputs:Optional[dict[str, Any|None]|Sequence[Any|None]] = None) -> None:
|
|
super().__init__(anp, key, XMLScrapTabModel, inputs)
|
|
|
|
def run(self:Self, actions:Sequence[Action], shared:dict[str, Any|None] = {}) -> dict[str, Any|None]:
|
|
|
|
self.do(Do(*actions), shared)
|
|
|
|
return shared
|
|
|
|
def __load(self:Self, url:str) -> Exception|None:
|
|
try:
|
|
|
|
response:Response
|
|
|
|
with get(url) as response:
|
|
self.tabs[self.tab_selected].set(response.text)
|
|
|
|
except Exception as exception:
|
|
return exception
|
|
return None
|
|
|
|
def execute(self:Self, action:Execute, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
|
try:
|
|
action.callback(shared)
|
|
action.done = True
|
|
except Exception as exception:
|
|
action.exception = exception
|
|
|
|
def open(self:Self, action:Open, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
|
|
|
url:str = shared[action.url] if action.url in shared and isinstance(shared[action.url], str) else action.url
|
|
|
|
self.tabs[action.key] = XMLScrapTabModel(action.key, url)
|
|
self.tab_selected = action.key
|
|
if url is not None:
|
|
action.exception = self.__load(url)
|
|
action.done = action.exception is None
|
|
|
|
def close(self:Self, action:Close, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
|
if action.key in self.tabs:
|
|
del self.tabs[action.key]
|
|
if len(self.tabs):
|
|
self.tab_selected = list(self.tabs.keys())[-1]
|
|
action.done = True
|
|
|
|
def switch(self:Self, action:Switch, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
|
if action.key in self.tabs:
|
|
self.tab_selected = action.key
|
|
action.done = True
|
|
|
|
def go(self:Self, action:Go, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
|
|
|
url:str = shared[action.url] if action.url in shared and isinstance(shared[action.url], str) else action.url
|
|
|
|
action.exception = self.__load(url)
|
|
action.done = action.exception is None
|
|
|
|
def get_selector(self:Self, selector:str, from_item:Optional[Selector] = None, force_multiple:bool = False) -> ParseSelectorList|ParseSelector|None:
|
|
if self.tab_selected in self.tabs:
|
|
|
|
path:ParseSelector = self.tabs[self.tab_selected].selector
|
|
items:ParseSelectorList = (
|
|
path.css(selector) if isinstance(selector, CSS) else
|
|
path.xpath(selector) if isinstance(selector, XPath) else
|
|
path.css(f"[name='{selector}']") if isinstance(selector, Name) else
|
|
path.css(f"a:contains('{selector}')") if isinstance(selector, LinkText) else
|
|
path.css(f".{selector}") if isinstance(selector, Class) else
|
|
path.css(f"#{selector}") if isinstance(selector, ID) else
|
|
path.css(selector))
|
|
|
|
return (
|
|
items if force_multiple or selector.multiple else
|
|
items[0] if len(items) else
|
|
None)
|
|
return [] if force_multiple or selector.multiple else None
|
|
|
|
def get(self:Self, action:Get, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
|
|
|
item:ParseSelector|ParseSelectorList|None = last_item if action.selector is None else self.get_selector(action.selector, last_item)
|
|
multiple:bool = False if action.selector is None else action.selector.multiple
|
|
attribute:str = "::text" if action.attribute == "CDATA" else f"::attr({action.attribute})"
|
|
|
|
if action.reset and action.key in shared:
|
|
del shared[action.key]
|
|
|
|
if multiple:
|
|
if action.key not in shared or not isinstance(shared[action.key], list):
|
|
shared[action.key] = []
|
|
shared[action.key].extend([subitem.css(attribute).get() for subitem in item])
|
|
else:
|
|
shared[action.key] = None if item is None else item.css(attribute).get()
|
|
|
|
def foreach(self:Self, action:ForEach, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
|
|
|
item:ParseSelector|Any|None
|
|
done:bool = True
|
|
|
|
for item in (
|
|
self.get_selector(action.items, last_item, True) if isinstance(action.items, Selector) else
|
|
action.items):
|
|
|
|
if not self.anp.working():
|
|
done = False
|
|
break
|
|
|
|
if action.callback is not None:
|
|
try:
|
|
item = action.callback(item)
|
|
except Exception as exception:
|
|
action.exception = exception
|
|
done = False
|
|
break
|
|
|
|
subactions:Do = Do(*action.actions)
|
|
|
|
self.do(subactions, shared, item if isinstance(item, ParseSelector) else last_item)
|
|
if not subactions.done or subactions.exception is not None:
|
|
action.exception = subactions.exception
|
|
done = False
|
|
break
|
|
|
|
action.done = done
|
|
|
|
def _and(self:Self, action:And, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> bool:
|
|
|
|
condition:Callable[[dict[str, Any|None]], bool]|Selector|Condition
|
|
|
|
action.ok = True
|
|
|
|
for condition in action.conditions:
|
|
if isinstance(condition, Condition):
|
|
self.do(condition, shared, last_item)
|
|
if not condition.ok:
|
|
action.ok = False
|
|
break
|
|
elif isinstance(condition, Selector):
|
|
if not self.get_selector(condition, last_item):
|
|
action.ok = False
|
|
break
|
|
elif callable(condition):
|
|
try:
|
|
if not condition(shared):
|
|
action.ok = False
|
|
break
|
|
except Exception as exception:
|
|
action.exception = exception
|
|
action.ok = False
|
|
break
|
|
|
|
def _or(self:Self, action:Or, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> bool:
|
|
|
|
condition:Callable[[dict[str, Any|None]], bool]|Selector|Condition
|
|
|
|
action.ok = False
|
|
|
|
for condition in action.conditions:
|
|
if isinstance(condition, Condition):
|
|
self.do(condition, shared, last_item)
|
|
if condition.ok:
|
|
action.ok = True
|
|
break
|
|
elif isinstance(condition, Selector):
|
|
if self.get_selector(condition, last_item):
|
|
action.ok = True
|
|
break
|
|
elif callable(condition):
|
|
try:
|
|
if condition(shared):
|
|
action.ok = True
|
|
break
|
|
except Exception as exception:
|
|
action.exception = exception
|
|
action.ok = False
|
|
break
|
|
|
|
def _not(self:Self, action:Not, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> bool:
|
|
if isinstance(action.condition, Condition):
|
|
self.do(action.condition, shared, last_item)
|
|
action.ok = not action.condition.ok
|
|
elif isinstance(action.condition, Selector):
|
|
action.ok = not self.get_selector(action.condition, last_item)
|
|
elif callable(action.condition):
|
|
try:
|
|
action.ok = not action.condition(shared)
|
|
except Exception as exception:
|
|
action.exception = exception
|
|
action.ok = False
|
|
|
|
def _if(self:Self, action:If, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
|
|
|
if not isinstance(action.condition, Condition):
|
|
action.condition = Condition((action.condition,))
|
|
|
|
self.do(action.condition, shared, last_item)
|
|
|
|
if action.condition.done and action.condition.exception is None:
|
|
|
|
subactions:Do = Do(*((action.if_actions if action.condition.ok else action.else_actions) or []))
|
|
|
|
self.do(subactions, shared, last_item)
|
|
action.done = subactions.done and subactions.exception is None
|
|
action.exception = subactions.exception
|
|
|
|
else:
|
|
action.done = action.condition.done
|
|
action.exception = action.condition.exception
|
|
|
|
def _while(self:Self, action:While, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
|
|
|
if not isinstance(action.condition, Condition):
|
|
action.condition = Condition((action.condition,))
|
|
|
|
while True:
|
|
if not self.anp.working():
|
|
action.done = False
|
|
break
|
|
|
|
self.do(action.condition, shared, last_item)
|
|
|
|
if not action.condition.done or action.condition.exception is not None:
|
|
action.done = action.condition.done
|
|
action.exception = action.condition.exception
|
|
break
|
|
|
|
if action.condition.ok:
|
|
|
|
subactions:Do = Do(*action.actions)
|
|
|
|
self.do(subactions, shared, last_item)
|
|
if not subactions.done or subactions.exception is not None:
|
|
action.exception = subactions.exception
|
|
action.done = False
|
|
break
|
|
|
|
else:
|
|
action.done = True
|
|
break
|
|
|
|
def do_while(self:Self, action:DoWhile, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
|
|
|
if not isinstance(action.condition, Condition):
|
|
action.condition = Condition((action.condition,))
|
|
|
|
while True:
|
|
if not self.anp.working():
|
|
action.done = False
|
|
break
|
|
|
|
subactions:Do = Do(*action.actions)
|
|
|
|
self.do(subactions, shared, last_item)
|
|
if not subactions.done or subactions.exception is not None:
|
|
action.exception = subactions.exception
|
|
action.done = False
|
|
break
|
|
|
|
self.do(action.condition, shared, last_item)
|
|
|
|
if not action.condition.done or action.condition.exception is not None:
|
|
action.done = action.condition.done
|
|
action.exception = action.condition.exception
|
|
break
|
|
|
|
if not action.condition.ok:
|
|
action.done = True
|
|
break
|
|
|
|
def download(self:Self, action:Download, shared:dict[str, Any|None] = {}, last_item:Optional[Any] = None) -> None:
|
|
if action.key or action.path:
|
|
try:
|
|
|
|
response:Response
|
|
url:str = shared[action.url] if action.url in shared and isinstance(shared[action.url], str) else action.url
|
|
|
|
with get(url) as response:
|
|
if response.content is not None:
|
|
if action.key:
|
|
if action.multiple:
|
|
if action.key not in shared or not isinstance(shared[action.key], list):
|
|
shared[action.key] = []
|
|
shared[action.key].append(response.content)
|
|
action.done = True
|
|
elif action.overwrite or action.key not in shared:
|
|
shared[action.key] = response.content
|
|
action.done = True
|
|
else:
|
|
if action.overwrite or not self.anp.files.exists(action.path):
|
|
self.anp.files.save(action.path, response.content)
|
|
action.done = True
|
|
|
|
except Exception as exception:
|
|
action.exception = exception |