Python 解析Access对象|极客教程

Python 解析Access对象，对于构成访问日志文件行数据的9个字段，此前创建的Access对象不会对其内部的一些元素进行分解。我们将通过整体分解将这些数据项分别解析为高层字段。单独执行这些解析操作可以简化每个处理阶段。这也让我们无须破坏日志分析的通用结构而替换整个处理过程中的一小部分。

下一阶段解析得到的对象AccessDetails是NamedTuple的一个子类，并且它封装了原始的Access元组。该对象使用如下一些额外的字段来分别表示解析细节：

from typing import NamedTuple, Optional
import datetime
import urllib.parse

class AccessDetails(NamedTuple):
    access: Access
    time: datetime.datetime
    method: str
    url: urllib.parse.ParseResult
    protocol: str
    referrer: urllib.parse.ParseResult
    agent: Optional[AgentDetails]

属性access是原始的Access对象，即简单字符串的集合。属性time是解析后的access.time字符串。属性method、url和protocol通过分解access.request字段得到。属性referrer是一个解析后的URL。

属性agent还可以分解为更细粒度的字段。非常规浏览器或网站爬虫会生成一个无法解析的agent字符串，因此该属性的类型提示为Optional。

以下属性构成了NamedTuple类的子类AgentDetails。

class AgentDetails(NamedTuple):
    product: str
    system: str
    platform_details_extensions: str

这些字段反映了描述代理最常用的语法。尽管这方面存在相当大的差异，但此特定子集似乎相当普遍。

以下3个解析器用于精细分解字段：

from typing import Tuple, Optional
import datetime
import re

def parse_request(request: str) -> Tuple[str, str, str]:
    words = request.split()
    return words[0], ' '.join(words[1:-1]), words[-1]

def parse_time(ts: str) -> datetime.datetime:
    return datetime.datetime.strptime(
        ts, "%d/%b/%Y:%H:%M:%S %z"
    )

agent_pat = re.compile(
    r"(?P<product>\S*?)\s+"
    r"\((?P<system>.*?)\)\s*"
    r"(?P<platform_details_extensions>.*)"
)

def parse_agent(user_agent: str) -> Optional[AgentDetails]:
    agent_match = agent_pat.match(user_agent)
    if agent_match:
        return AgentDetails(**agent_match.groupdict())
    return None

我们为HTTP请求、时间戳和用户代理信息编写了3个解析器。日志中的请求值通常是由3个词构成的字符串，如GET /some/path HTTP/1.1。函数parse_request()提取了这3个以空格分隔的值。如果路径中也包含空格，那么提取第一个词和最后一个词分别作为方法和协议，剩下的词则作为路径的一部分。

时间解析委派给了datetime模块，并且parse_time()函数中提供了适当的格式。

解析用户代理比较有挑战性。对此有许多方法，我们为parse_agent()函数选择了一种常用的处理方法。如果用户代理文本与给定的正则表达式匹配，那么使用AgentDetails类的属性。如果用户代理信息与正则表达式不匹配，那么使用None值代替。原始文本可以从Access对象中获取。

基于给定的Access对象，我们将使用这3个解析器构建AccessDetails实例。函数access_detail_iter()的主体如下所示：

from typing import Iterable, Iterator
def access_detail_iter(access_iter: Iterable[Access]) -> Iterator[AccessDetails]:
    for access in access_iter:
        try:
            meth, url, protocol = parse_request(access.request)
            yield AccessDetails(
                access=access,
                time=parse_time(access.time),
                method=meth,
                url=urllib.parse.urlparse(url),
                protocol=protocol,
                referrer=urllib.parse.urlparse(access.referer),
                agent=parse_agent(access.user_agent)
            )
        except ValueError as e:
            print(e, repr(access))

我们使用了与先前access_iter()函数类似的设计模式。解析某个输入对象，基于结果构建了一个新对象。新的AccessDetails对象会封装之前的Access对象。利用这种技术让我们不仅可以使用不可变对象，还能包含更多详细信息。

这个函数本质上是从一个Access对象到一个AccessDetails对象的映射。另一种替代设计如下，它使用了相对高阶的map()函数。

from typing import Iterable, Iterator
def access_detail_iter2(
        access_iter: Iterable[Access]
    ) -> Iterator[AccessDetails]:

    def access_detail_builder(access: Access) -> Optional[AccessDetails]:
        try:
            meth, uri, protocol = parse_request(access.request)
            return AccessDetails(
                access=access,
                time=parse_time(access.time),
                method=meth,
                url=urllib.parse.urlparse(uri),
                protocol=protocol,
                referrer=urllib.parse.urlparse(access.referer),
                agent=parse_agent(access.user_agent)
            )
        except ValueError as e:
            print(e, repr(access))
        return None

    return filter(
        None,
        map(access_detail_builder, access_iter)
    )

我们从构造AccessDetails对象改为构造一个返回单个值的函数。可以将该函数映射到原始Access对象的可迭代输入流中，这与multiprocessing模块的工作方式非常相适。

在面向对象编程环境中，这些额外的解析器可以是类定义中的方法函数或者属性。具有惰性解析方法的面向对象设计的优点是数据项只在需要时才会被解析。这个特定的函数式设计方法可以解析任何东西，只需假定会用到它。

创建一个惰性函数式设计是可行的，它可以基于3个解析器函数并根据需求从Access对象中提取并解析各种元素。我们使用parse_time(access.time)参数而不使用details.time属性。尽管语法上变长了，但它确保了只在需要时才解析属性。