Leetao's Blog

Talk is cheap, show me the code

0%

发表于 更新于

分享一个很久之前写的一个关于滴滴行程的脚本，脚本的主要功能是根据滴滴行程分享的 url 获取订单情况，通过定时轮询还可以获取乘客到目的地的距离和时间

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
"""
@File    :   didi.py    
@Description：滴滴行程分享 Api
@Modify Time      @Author    @Version    @Description
------------      -------    --------    -----------
2021/5/29 8:47 上午   leetao      1.0         None
"""
import requests
import urllib.parse as urlparse
from urllib.parse import parse_qs
from dataclasses import dataclass, asdict

_header = {
    'Accept': 'application/json, text/plain, */*',
    'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.77 Safari/537.36',
    'Accept-Language': 'zh-CN,zh;q=0.9'
}


@dataclass
class DidiOrderParams:
    uid: str
    oid: str
    sign: str
    productType: str
    carProductid: str


def is_didi(url: str) -> bool:
    return url.startswith("https://z.didi.cn")


def get_order_params(url: str) -> DidiOrderParams:
    resp = requests.get(url, allow_redirects=False, headers=_header)
    if resp.status_code == 302:
        redirect_url = resp.headers['location']
        parsed = urlparse.urlparse(redirect_url)
        params = parse_qs(parsed.query)
        return DidiOrderParams(uid=params['uid'], oid=params['oid'], sign=params['sign'],
                               productType=params['productType'], carProductid=params['carProductid'])
    return None


def get_order_info(order_param: DidiOrderParams) -> str:
    resp = requests.get('https://common.diditaxi.com.cn/webapp/sharetrips/page/getOrderStatus',
                        params=asdict(order_param), headers=_header)
    assert resp.status_code == 200
    resp_json = resp.json()
    assert resp_json['errno'] == 0
    return f'本次行程 \n 起点：{resp_json["fromAddress"]} \n 终点: {resp_json["toAddress"]} \n 您的行程管家将定时为您播报行程情况'


def get_driver_loc(order_param: DidiOrderParams):
    resp = requests.get('https://common.diditaxi.com.cn/webapp/sharetrips/page/getDriverLoc',params=asdict(order_param),
                        headers=_header)
    assert resp.status_code == 200
    resp_json = resp.json()
    assert resp_json['errno'] == 0
    if 'etaDistance' not in resp_json:
        return f'本次行程已经结束'
    return f'距离目的地还有：{resp_json["etaDistance"]} 公里，预计还有：{resp_json["etaTime"]} 分钟'

发表于 更新于

要不要换个新的城市开始新的生活？

这是今年一直盘旋在我脑海里的想法。想法到落地中间并不是简单的 1 + 1 的问题，俩者之间隔着巨大的鸿沟，因为这意味着可能一切都需要重新开始。

从零开始并不是想象的那么简单，需要离开从大学开始就一直生活的城市，一个对我来说生活了八年多的地方，离开熟悉的天空、熟悉的街道、熟悉的人和熟悉的工作环境。记得上次有这种想法的时候，是刚毕业一年多的时候，也就是四年前，很遗憾，那个时候一个人的勇气并不能够支撑着我背起行囊潇洒地离开这里。

是的，没错，这次离开，是两个人一起。两个人能够在这种事情上达成一致的想法，是何其幸运的事情。在这里这些年最大的收获，就是收获了爱情，在开始新的阶段的时候，有人与我同行。

勇气是打开新的阶段的钥匙的话，那么找工作就是到达终点前的“拦路虎”了。

由于自己还处于在职状态，所以所有的面试基本上都只能约在下班之后了，自然面试前都准备工作也同样只能放在下班之后了。人生最让人觉得（有趣）无奈的事情就是，你越不想事情多的时候，基本上你的事情就会越来越多。恰巧开始准备复习的时候，负责的项目需要演示汇报，而这个时候郑州又赶上了暴雨加疫情，所谓的“涝疫结核”，一个多月的足不出户的生活，还有爷爷的离开。

因为疫情的原因，有一两年没有回老家了，没想到再次回去就是收到爷爷病重的消息，紧接着就是在暴雨和疫情期间被告知爷爷离开的消息，我记得那天的没有下雨，天比较阴沉，我坐在沙发上接到老爸打来的电话，说的什么我已经记不清了，我只记得挂断电话之后，眼泪不由自主地流了下来。

那个时候，我才切身体会道什么叫做紧绷着的弦，所有的事情都在不停地拉扯着神经。整个人变得有点焦虑，焦虑到夜里睡不着，焦虑到一个多月瘦了6-7斤，焦虑到茶饭不思。很难想象如果这个时候没有我的另一半一直陪着我，我该如何度过这一阶段。

潮起日落，日子总是一天一天在指尖划过。度过最难过的一段时光，后面的面试除了时间上比较紧凑，面试后面还是比较顺利的。工作了五年之后，再次收获了一份满意的 offer，很快就要去新的城市开始新的生活了。你说我现在的心情是什么样的，有向往，也有紧张，向往全新的生活，对新的开始难免有点彷徨和紧张。

最近处于离职的阶段，也算是闲了下来，可以偶尔跑跑步、看看书，前几天晚上还把藏在柜子里深处的尤克里里翻了出来，久违的轻松惬意，希望 2021 年接下来的日子都顺顺利利的。

发表于 更新于

原文连接： When to use assert

前言

assert 又称为断言，在 Python 代码中经常被使用，但是显然也存在滥用的情况。那么在什么时候使用 assert 呢？又或者 assert 的最佳实践是怎么样的呢？

assert 的使用

Python 的 assert 通常用来检查一个条件，如果它是真的，则不做任何事情，如果它是假的，则引发一个 AssertionError，并给出一个可选的错误信息:

1
2
3
4
5
6
py> x = 23
py> assert x > 0, "x is not zero or negative"
py> assert x%2 == 0, "x is not an even number"
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
AssertionError: x is not an even number

许多人喜欢通过 assert 来简单地快速地触发异常，比如下面的代码：

1
2
if not isinstance(x, int):
    raise AssertionError("not an int")

通过检验参数，并抛出 AssertionError ，实际上，这种做法是错误的，并且还很危险。正确的做法是应该抛出一个 TypeError 。

之所以危险是英文，assert 有一个特性：使用 -O 或 -OO 优化指令去运行 Python 的话，它会被被编译掉，而永远不会被执行。当你能够正确使用 assert 时，这是会是一个 feature，但当使用不当的话，则会导致代码在使用 -O 标志运行时出问题。

所以应该在什么情况下使用 assert 呢

什么时候使用 assert

在下面这四种情况下都可以考虑使用 assert:

  • 防御性编程。
  • 对程序逻辑进行运行时检查。
  • 契约检查（如前条件和后条件）。
  • 程序不变性
  • 检查文件

对断言的看法各不相同，它可以看作是一种对代码正确性的信心的表现形式吧。如果你确定你的代码毫无问题的话，显然断言（assert）是没有任何意义的，完全可以把这些断言移除；如果你确定断言可能会被触发，那你也完全可以不用断言（assert），毕竟它在某些情况下肯定会编译掉，导致你的检查被跳过。

介于这两种情况之间的情况才是有趣的，当你确定代码是正确的，但又不完全确定的时候。
你确定代码是正确的，但又不是绝对确定。有可能漏掉一些场景，在这种情况下，通过断言（assert）进行运行时检查有助于我们尽早的发现错误。

断言的另一个用途是检查程序的不变量。所谓的不变量就是一些你可以依赖的真实条件，除非一个错误导致它变成假的。如果有一个bug，最好能尽早发现所以我们要对它进行测试，但我们不想因为这样的测试而降低代码的速度。但我们不想因为这些测试而使代码变慢。因此我们可以通过可在开发中打开而在生产中关闭断言（assert）。

不变量的一个例子是，如果你的函数预期一个数据库连接是 open，并承诺它在返回时仍然是open，这就是函数的一个不变式。open 就是该函数的一个不变量。

1
2
3
4
5
def some_function(arg):
    assert not DB.closed()
    ... # code goes here
    assert not DB.closed()
    return result

断言也是很好的检查性评论，而不是写一个评论：

1
2
3
# when we reach here, we know that n > 2
# 我们可以通过将其转换为断言来确保在运行时对其进行检查
assert n > 2

断言也是防御性编程的一种形式。你不是在保护现在的代码中的错误，而是在保护以后引入错误的变化。理想情况下，单元测试会发现这些错误，但是实际上，即使存在测试，它们也是往往是不完整的。有可能在几周内都没有人注意到构建机器人，或者有的时候在提交代码前忘记运行测试。有一个内部检查是防止错误潜入的另一道防线，特别是阻止那些会导致代码故障和返回错误结果的静默错误。

假设你有个代码中有 if…elif 代码块，这个时候你是知道这些分支对应了哪些变量，并且了解逻辑是什么:

1
2
3
4
5
6
7
# target is expected to be one of x, y, or z, and nothing else.
if target == x:
    run_x_code()
elif target == y:
    run_y_code()
else:
    run_z_code()

假设这个代码现在是完全正确的。但它会一直正确吗？需求改变了,代码也会改变。现在要求改变为当 target == w，执行 run_w_code。如果我们改变了设置目标的代码，但忽略了改变这个代码块，它将错误地调用 run_z_code()：

1
2
3
4
5
6
7
8
target = w
# ...
if target == x:
    run_x_code()
elif target == y:
    run_y_code()
else:
    run_z_code()

最好是把这段代码写成防御性的，哪怕之后有变化，它也是要么是正确的，要么立即失败。

写代码块一开始加注释是很好的习惯，但是通常时间久了，我们就没有阅读和更新注释的习惯。有可能注释很快就会被淘汰。但通过一个断言，我们既可以记录这个块的假设，也可以在断言检验没通过的时候直接抛出错误。

1
2
3
4
5
6
7
8
assert target in (x, y, z)
if target == x:
    run_x_code()
elif target == y:
    run_y_code()
else:
    assert target == z
    run_z_code()

断言既是防御性编程，也是经过检查的文档，下面这个代码会更好一点：

1
2
3
4
5
6
7
8
9
if target == x:
    run_x_code()
elif target == y:
    run_y_code()
elif target == z:
    run_z_code()
else:
    # This can never happen. But just in case it does...
    raise RuntimeError("an unexpected error occurred")

契约设计是另一种对断言的不错的应用。在按契约设计中，我们认为函数与它们的调用者签订了 “契约”。
例如,”如果你传递给我一个非空的字符串，我保证返回该字符串的第一个的第一个字符转换成大写字母”。

如果这个契约被函数或调用它的代码所破坏,代码就是有问题的。我们说，函数有预设条件（即
参数的约束）和后置条件（返回结果的约束） 因此，这个函数可能被代码为：

1
2
3
4
5
6
7
def first_upper(astring):
    assert isinstance(astring, str) and len(astring) > 0
    result = astring[0].upper()
    assert isinstance(result, str) and len(result) == 1
    assert result == result.upper()
    return result

契约设计的目的是，在一个正确的程序中，前条件和后条件总是成立的。当我们发布无错误的程序并将其投入生产时，我们可以安全地删除这些断言。

什么时候不使用assert

  • 永远不要用它们来测试用户提供的数据，或者用于在任何情况下必须进行检查的地方。
  • 不要用断言来检查任何你认为在你的程序的正常使用中可能失败的东西。断言是为特殊失败条件。你的用户不应该看到 AssertionError。如果他们看到了，这就是一个需要修复的错误。
  • 特别是，不要因为 assert 比一个测试和 raise 短就使用它。
  • 不要用它们来检查公共库的输入参数函数的输入参数，因为你无法控制调用者，也不能保证它不会破坏函数的契约。
  • 不要将 assert 用于任何你期望恢复的错误。 换句话说，你没有理由在生产代码中捕捉一个AssertionError 异常。
  • 不要使用太多的断言，以至于它们掩盖了代码。

发表于 更新于

前言

Python3.7 引入了一个新的模块那就是 dataclasses，早在 3.6 版本的时候我就通过安装 dataclasses 三方库体验了一波，那么为什么要用 dataclasses 呢？

为什么使用 dataclasses

一个简单的场景，当你想定义一个对象的属性的时候，比如一本书，通常你会这样

1
2
3
4
5
class Book:
    def __init__(self, name: str, price: float, author:str = "佚名"):
        self.name = name
        self.price = price
        self.author = author

如果在不定义 __repr__ 的情况下，初始化这个对象，并输出的话:

1
2
3
4
5
6
7
>>> book = Book('桃子',10.0,author='桃子')
>>> book
<__main__.Book object at 0x10b69cbb0>
>>> str(book)
'<__main__.Book object at 0x10b69cbb0>'
>>> repr(book)
'<__main__.Book object at 0x10b69cbb0>'

显然输出不够友好，对于属性比较少的对象，定义一个 __repr__ 并不太麻烦，但是一旦需要定义的对象很多或者属性很多的情况下，这样子做就显得略微麻烦了一点。这个时候 dataclasses 就派上用场了。

dataclasses

我们用 dataclasses 把 Book 改造一下

1
2
3
4
5
6
7
from dataclasses import dataclass

@dataclass
class Book:
	name:str
	price:float
	author:str = '桃子'

然后简单验证一下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
>>> from dataclasses import dataclass,asdict,astuple
>>> @dataclass
... class Book:
...     name:str
...     price:float
...     author:str
...
>>> book1 = Book(name='桃子1',price=10.0,author='桃子1')
>>> book1
Book(name='桃子1', price=10.0, author='桃子1')
>>> asdict(book1)
{'name': '桃子1', 'price': 10.0, 'author': '桃子1'}
>>> astuple(b1)
('桃子1', 12.0, '桃子')

dataclasses 甚至还具备 asdict 函数可以将对象转成 dict，也存在 astuple 可以将对象转成tupple ,是不是很方便,但是还不够，有时候我们对不同对参数进行一定对校验，很遗憾 dataclasses 并不能做到，这个时候就需要看 attrs 和 pydantic 了。

除此之外，attrs 和 pydantic 还有其他的 dataclasses 不具备的特性，见下表：

attrs vs pydantic

attrs 和 pydantic 都需要通过 pip 安装

1
2
pip install attrs
pip install pydantic

validator

attrs 中的 validators

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import attr

@attr.s
class Book:
    name: str = attr.ib(default=None)
    price:float = attr.ib(default=None,validator=attr.validators.instance_of(float))
    author: str = attr.ib(default='桃子')

    @price.validator
    def more_than_zero(self, attribute, value):
        if not value > 0:
            raise ValueError("价格不能少于 0 元!")

book = Book(name='attrs',price=-1.0,author='hynek Hynek Schlawack')

运行一下，验证一下结果：

1
2
3
4
5
6
7
8
9
Traceback (most recent call last):
  File "test_attrs.py", line 14, in <module>
    book = Book(name='attrs',price=-1.0,author='hynek Hynek Schlawack')
  File "<attrs generated init __main__.Book>", line 6, in __init__
  File "D:\workspace\python\test\venv\lib\site-packages\attr\_make.py", line 2975, in __call__
    v(inst, attr, value)
  File "test_attrs.py", line 12, in more_than_zero
    raise ValueError("价格不能少于 0 元!")
ValueError: 价格不能少于 0 元!

pydantic 中的 validators

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from pydantic import BaseModel,validator, ValidationError

class Book(BaseModel):
    name:str
    price:float
    author:str = '桃子'

    @validator("price")
    def more_than_zero(cls, v):
        assert isinstance(v,float)
        if v <= 0:
            raise ValueError("价格不能少于 0 元!")
        return v

book = Book(name='attrs',price=-1.0,author='hynek Hynek Schlawack')

同样的验证一下结果:

1
2
3
4
5
6
7
Traceback (most recent call last):
  File "test_pydantic.py", line 15, in <module>
    book = Book(name='attrs',price=-1.0,author='hynek Hynek Schlawack')
  File "pydantic\main.py", line 406, in pydantic.main.BaseModel.__init__
pydantic.error_wrappers.ValidationError: 1 validation error for Book
price
  价格不能少于 0 元! (type=value_error)

除了 validators ，attrs 和 pydantic 还同时具有 converters（转换器） 功能。

converters

转换器也是一个十分实用的功能，所谓的转换器就是在将参数在传递到 __init__ 之前，将它按照需求转换成所需格式的数据。

最常见的就是将 str 转成 datetime 格式，attrs 和 pydantic 在转换器的实现上有所区别：

attrs 中 的转换器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import attr
from datetime import datetime

def conver_to_datetime(user_input) ->  datetime:
    if isinstance(user_input,str):
        return datetime.strptime(user_input, '%Y-%m-%d %H:%M:%S')
    elif isinstance(user_input,datetime):
        return user_input
    raise TypeError("不支持的参数类型，仅支持:str 和 datetime")

@attr.s
class Book:
    # ... 省略
    publish_date: datetime = attr.ib(default=None, converter=conver_to_datetime)

book = Book(name='attrs',price=1.0,author='hynek Hynek Schlawack',publish_date='2021-09-13 00:00:00')
print(book)

运行程序

1
Book(name='attrs', price=1.0, author='hynek Hynek Schlawack', publish_date=datetime.datetime(2021, 9, 13, 0, 0))

不难看出，str 类型的字符串被成功转化了。

pydantic 中的转换器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from pydantic import BaseModel,validator
from datetime import datetime

class Book(BaseModel):
		# 省略...
    publish_date: datetime

    @validator("publish_date")
    def conver_to_datetime(cls, user_input):
        if isinstance(user_input,str):
            return datetime.strptime(user_input, '%Y-%m-%d %H:%M:%S')
        elif isinstance(user_input,datetime):
            return user_input
        raise TypeError("不支持的参数类型，仅支持:str 和 datetime")

book = Book(name='attrs',price=1.0,author='hynek Hynek Schlawack', publish_date='2021-09-13 16:03:00')
print(book)

验证结果:

1
name='attrs' price=1.0 author='hynek Hynek Schlawack' publish_date=datetime.datetime(2021, 9, 13, 16, 3)

从上面的代码不难看出，attrs 本身内置 converter 参数，可以通过传递内置函数或者自定义函数实现转化器的功能，而 pydantic 则需要借助 validator 去实现，在这一点上 pydantic 还是稍微逊色 attrs。

除了上面提到的两个功能，attrs 和 pydantic 都具备 immutable 的方法用来修饰属性，从而实现属性不可修改。

attrs 还具备 slots 和 programmatic creation 有兴趣的可以进一步阅读官方文档。

总结

从上面的例子，不难看出 pydantic 有下面几个问题：

  1. pydantic 不支持位置参数
  2. pydantic 的输出有点奇怪，没有带上类名
  3. pydantic 不支持 slots 和 programmatic creation
  4. pydantic 不支持 Collection 类型

在参考文章中还提到了 pydantic 对 unions 的策略有问题，不容易定制，并且对定制的（非）结构化的支持很弱。

所以如果有复杂的需求的话，建议使用 attrs ，只是想简单的呈现对象的属性的话，可以考虑用 dataclasses 。

参考文章

dataclasses vs attrs vs Pydantic

attrs 和 Python3.7 的 dataclasses

Why I use attrs instead of pydantic

发表于 更新于

前言

默认情况下 es 是不支持 pdf、doc 等文档的搜索的，但是可以通过安装 Ingest attachment plugin 插件来使得 es 提取通用格式的文件，从而可以实现搜索的功能。

安装与使用

安装 Ingest attachment plugin

安装很简单，通过 elasticsearch-plugin 可以直接进行安装

1
bin/elasticsearch-plugin install ingest-attachment

Ingest attachment plugin 允许 Elasticsearch 通过使用 Apache 文本提取库 Tika 提取通用格式（例如：PPT，XLS 和 PDF）的文件附件。Apache Tika 工具包可从一千多种不同的文件类型中检测并提取元数据和文本。所有这些文件类型都可以通过一个界面进行解析，从而使 Tika 对搜索引擎索引，内容分析，翻译等有用。需要注意的是，源字段必须是 Base64 编码的二进制，如果不想增加在 Base64 之间来回转换的开销，则可以使用 CBOR 格式而不是 JSON，并将字段指定为字节数组而不是字符串表示形式，这样处理器将跳过 Base64 解码。

创建 attachment pipeline

通过 kibana 的开发工具进行请求

1
2
3
4
5
6
7
8
9
10
11
PUT _ingest/pipeline/pdfattachment
{
  "description": "Extract attachment information encoded in Base64 with UTF-8 charset",
  "processors": [
    {
      "attachment": {
        "field": "file"
      }
    }
  ]
}

返回结果：

1
2
3
{
  "acknowledged" : true
}

表示创建成功，接下来就是验证上传 pdf 以及搜索功能了。

转换并上传PDF文件的内容到Elasticsearch中

对于 Ingest attachment plugin 来说，它的数据必须是 Base64 的。这里为了快速创建，我们通过一个 bash 脚本去处理用来测试的 pdf

1
2
3
4
5
6
7
8
9
!/bin/bash

encodedPdf=`cat sample.pdf | base64`

json="{\"file\":\"${encodedPdf}\"}"

echo "$json" > json.file

curl -X POST 'http://localhost:9200/pdf-test1/_doc?pipeline=pdfattachment&pretty' -H 'Content-Type: application/json' -d @json.file

上传成功会返回如下的结果:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
{
  "_index" : "pdf-test1",
  "_type" : "_doc",
  "_id" : "1oVSxHsB1ubIHqCXIbPU",
  "_version" : 1,
  "result" : "created",
  "_shards" : {
    "total" : 2,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 0,
  "_primary_term" : 1
}

如果出现提示下述提示

1
2
3
4
5
{
	"statusCode": 400,
	"error": "Bad Request",
	"message": "Payload content length greater than maximum allowed: 1048576"
}

说明上传的文件大小超过了 Kibana 默认的上限（默认 1M），修改 kibana 的配置 kibana.yml

1
server.maxPayloadBytes: "209715200"

查看索引并搜索

查看索引

通过下面的命令可以查看 pdf-test1 的索引情况:

1
GET pdf-test1/_search

结果如下图

其中 _source 里有个 content 字段，就是 pdf 的内容，结果如下:

1
2
3
4
5
6
7
8
9
10
11
12
{
  "_source": {
		 "file":"...",
			"attachment": {
                        "date": "2021-08-18T07:29:34Z",
                        "content_type": "application/pdf",
                        "author": "author",
                        "language": "lt",
                        "title": "2021-07(copy)",
                        "content": "..."
   }
}

其中 file 就是 base64 格式的内容，content 则包含了 pdf 的内容，如果不想要 file 则可以通过 remove processor 去除这个字段

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
PUT _ingest/pipeline/pdfattachment
{
  "description": "Extract attachment information encoded in Base64 with UTF-8 charset",
  "processors": [
    {
      "attachment": {
        "field": "file"
      }
    },
    {
      "remove": {
        "field": "file"
      }
    }
  ]
}

搜索

通过在 kibana 的开发工具执行下述命令，验证一下搜索：

1
2
3
4
5
6
7
8
GET pdf-test1/_search
{
  "query": {
    "match": {
      "attachment.content": "5G"
    }
  }
}

参考链接

Elasticsearch：如何对PDF文件进行搜索

发表于 更新于

安装启动 Elasticsearch

1
2
3
docker network create elastic
docker pull docker.elastic.co/elasticsearch/elasticsearch:7.14.0
docker run -d --name es01-test --net elastic -p 9200:9200 -p 9300:9300 -e "discovery.type=single-node" docker.elastic.co/elasticsearch/elasticsearch:7.14.0

安装启动 Kibana

1
2
docker pull docker.elastic.co/kibana/kibana:7.14.0
docker run -d --name kib01-test --net elastic -p 5601:5601 -e "ELASTICSEARCH_HOSTS=http://es01-test:9200" docker.elastic.co/kibana/kibana:7.14.0

启动完成之后访问 http://localhost:5601 即可，默认是英文，配置中文可以参考

Kibana 汉化

在 kibana.yml 中加入

1
i18n.locale: "zh-CN"

然后重启 kibana 即可

配置完成后重启服务，界面如下图：

安装 IK 中文分词

es 中本身自带英文分词，但是实际业务需要存在中文分词的场景，所以手动安装中文分词插件。

下载 IK 分词插件

从 github 上根据不同的 es 版本下载对应的 ik 插件压缩包

https://github.com/medcl/elasticsearch-analysis-ik/releases

在 es 的 plugins 解压

1
2
cd your-es-plugins/plugins/ && mkdir ik
unzip elasticsearch-analysis-ik-7.14.0.zip

然后重新启动 es，这里我是使用 docker 启动 es 的，所以先通过 docker cp 的命令将插件复制到容器中的

1
docker cp es.zip container-id:/usr/share/elasticsearch/plugins/es

重启 es

重启 es 的服务，如果成功加载插件，控制台会有如下的输出:

发表于 更新于

Recently I was working on rss3 SDK. In order to facilitate development, I just make a python version of the reference JavaScript SDK, which means the usage should be pretty similar between both.

Make a Python version

What’s more, to make a clear code,I use type hinting in the new project. However, there is a code snippet in JavaScript version:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
# index.ts
class RSS3 {
    options: IOptions;
    persona: Persona;
    file: File;
    profile: Profile;
    items: Items;
    item: Item;
    links: Links;

    constructor(options: IOptions) {
        this.options = options;

        this.file = new File(this);
        this.persona = new Persona(this);
        this.profile = new Profile(this);
        this.items = new Items(this);
        this.item = new Item(this);
        this.links = new Links(this);
    }
}
# file.ts
import Main from './index';
class File {
    private main: Main;
    private list: {
        [key: string]: RSS3IContent;
    } = {};
    private dirtyList: {
        [key: string]: number;
    } = {};

    constructor(main: Main) {
        this.main = main;
    }
...
}

Yes, the two files refer to each other. Actually, it happens on many other files. When I turned this into Python version:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# index.py
class RSS3:
		...
    persona: Persona
    file: File_
    profile: Profile
    items: Items
    items: Item
    links: Links

# file.py
from .index import RSS3

class File:
    rss3: RSS3

def __init__(self, rss3: RSS3):
        self.rss3 = rss3

circular import

It didn’t seem any errors, but when I started testing the problems appeared.

1
E   ImportError: cannot import name 'RSS3' from partially initialized module 'rss3.src.index' (most likely due to a circular import)

how to solve this problem ? Don’t worry, PEP 484 has given a solution.

Solutions

When a type hint contains names that have not been defined yet, that definition may be expressed as a string literal, to be resolved later.

So we can modify our code like the following:

1
2
3
4
5
class File:
    rss3: 'RSS3'

def __init__(self, rss3: 'RSS3'):
        self.rss3 = rss3

That’s okay already.

Sometimes there’s code that must be seen by a type checker (or other static analysis tools) but should not be executed. For such situations the typing module defines a constant, TYPE_CHECKING, that is considered True during type checking (or other static analysis) but False at runtime.

Modify out code again:

1
2
3
4
5
6
7
8
if TYPE_CHECKING:
    from .index import RSS3

class File:
    rss3: 'RSS3'

	def __init__(self, rss3: 'RSS3'):
	        self.rss3 = rss3

What’s more

if we are using Python 3.7+, we can at least skip having to provide an explicit string annotation by taking advantage of PEP 563:

1
2
3
4
5
6
7
8
9
10
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
    from .index import RSS3

class File:
    rss3: 'RSS3'

	def __init__(self, rss3: 'RSS3'):
	        self.rss3 = rss3

The from __future__ import annotations import will make all type hints be strings and skip evaluating them.

References

PEP 484 – Type Hints

Python type hinting without cyclic imports

发表于 更新于

前言

前段时间用 Flutter 做了一个开源的项目 RSSAid,因为需要打包 apk，在此之前一直是在本地签名打包的。后来和别人交流了一下，想起来可以用 Github Action 构建持续化集成，自动打包。然后就研究了一下，最后完成了根据 tag 版本自动生成 apk 的 workflows。

Workflows

自动化构建脚本如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
# main.yml
# 自动构建 Apk
name: Test, Build and Release apk

# 工作流程触发的时机，这里是当一个版本标签推送到仓库时触发
on:
  push:
    tags:
      - v*

# 这个工作流程需要执行的任务
jobs:
  process:
    name: all process
    runs-on: ubuntu-latest
    # 这个任务的步骤
    steps:
      # 拉取项目代码
      - uses: actions/checkout@v2
      # 建立 java 环境
      - name: Setup Java JDK
        uses: actions/setup-java@v1.4.3
        with:
          java-version: "12.x"
      # 建立 Flutter 环境
      - name: Flutter action
        uses: subosito/flutter-action@v1.4.0
        with:
          channel: "stable"
          flutter-version: "1.22.4"
      # 下载项目依赖
      - run: flutter pub get
      - run: echo $ENCODED_KEYSTORE | base64 -di > android/app/keystore.jks
        env:
          ENCODED_KEYSTORE: ${{ secrets.ENCODED_KEYSTORE }}
      # 打包 APK
      - run: flutter build apk --release
        env:
          KEYSTORE_PASSWORD: ${{ secrets.KEYSTORE_PASSWORD }}
          KEY_ALIAS: ${{ secrets.KEY_ALIAS }}
          KEY_PASSWORD: ${{ secrets.KEY_PASSWORD}}
      # 发布到 Release
      - name: Release apk
        uses: ncipollo/release-action@v1.5.0
        with:
          artifacts: "build/app/outputs/apk/release/*.apk"
          token: ${{ secrets.RELEASE_TOKEN }}

使用注意事项

脚本中有很多环境变量，都需要实现定义好，在项目的 secrets 中添加上。

RELASE_TOKEN

这个环境变量需要在 Personal access tokens 申请，需要注意的是，申请完成之后，不要着急关闭这个页面，因为一旦关闭就不能再次查看生成的 token 了，这个 token 需要申请 repo 和 workflow 的权限

生成 token 成功后，找到项目的 **Settings => Secrets **选项，新建名为 RELEASE_TOKEN 的 secrets 然后 value 值为刚才生成的 token。这个完成之后，就需要对设置生成 apk 需要的签名进行变量设置了。

签名相关变量

正常 app 签名步骤可以参考 app签名，最终我们会创建一个 key.properties 的文件，文件内容如下：

1
2
3
4
storePassword=<password from previous step>
keyPassword=<password from previous step>
keyAlias=key
storeFile=<location of the key store file, e.g. /Users/<user name>/key.jks>

然后在 android/app/build.grade 中配置

1
2
3
4
5
6
7
8
signingConfigs {
    release {
        keyAlias keystoreProperties['keyAlias']
        keyPassword keystoreProperties['keyPassword']
        storeFile keystoreProperties['storeFile'] ? file(keystoreProperties['storeFile']) : null
        storePassword keystoreProperties['storePassword']
    }
}

但是将 key.properties 上传到仓库显然是不安全的。所以需要对代码进行修改，将对应的变量添加到 secrets 中，从 secrets 中获取变量。

1
2
3
4
storeFile file(System.getenv("KEYSTORE") ?:"keystore.jks")
storePassword System.getenv("KEYSTORE_PASSWORD")
keyAlias System.getenv("KEY_ALIAS")
keyPassword System.getenv("KEY_PASSWORD")

其中 **KEYSTORE_PASSWORD、KEY_ALIAS、KEY_PASSWORD **直接就可以添加，那么 **KEYSTORE **这个变量怎么处理呢？KEYSTORE 对应着 jks 文件位置。jks 文件显然也不可能上传到仓库，所以我们换种方法，在构建的时候生成 jks 文件。

构建时生成 jks 文件

正常情况下打开生成的 jks 文件多半是乱码，所以我们可以通过 base64 对文件进行编码，然后在构建的时候，再解码重新生成文件。

获取 base64 格式的 keystore

首先获取 base64 格式的 keystore 

1
openssl base64 -A -in <jks.文件位置>

然后将输出的结果复制下来

保存 base64 格式的 keystore

将编码后的 keystore 内容，添加到 secrets ，变量名命名为 ENCODED_KEYSTORE，然后在构建过程中就可以将 keystore 文件还原了。

1
echo $ENCODED_KEYSTORE | base64 -di > android/app/keystore.jks

发表于 更新于

前言

主备同步，也叫主从复制，是 mysql 提供的一种高可用的解决方案，保证主备数据一致性的解决方案。
在生产环境中，会有很多不可控因素，比如数据库服务器宕机等，因此在生产环境中，都会采用主备同步。在应用的规模不大的情况下，一般会采用一主一备。除此之外，采用主备同步还可以：

  • 提高数据库的读并发性，大多数应用都是读比写要多，采用主备同步方案，当使用规模越来越大的时候，可以扩展备库来提升读能力。
  • 备份，主备同步可以得到一份实时的完整的备份数据库。
  • 快速恢复，当主库出错了（比如误删表），通过备库来快速恢复数据。

那么主备同步的原理是什么？

主备同步的实现原理

主备同步模式之所以能够实现，显然是有一种手段可以，将主的 mysql 服务（以下简称 master）执行的 DDL 和 DML 语句传递给 备份的 mysql 服务（以下简称为 slave），这个就是 MySQL 的 binlog，mysql 的 binlog 是 MySQL 最重要的日志，它记录了所有的DDL和DML(除了数据查询语句)语句，以事件形式记录，还包含语句所执行的消耗的时间，MySQL的二进制日志是事务安全型的。
所以 binlog 一般会有两个用途：

  • 主备模式下，master 将 binlog 传递给 slave，从而达到数据一致的目的
  • 数据恢复，可以通过 mysqlbinlog 工具来恢复数据

当 master 将 binlog 传递到 slave 的时候，会被传到 slave 的 relay log，relay log 也叫中继日志，是连接 master 和 slave 的核心，relay-log 的结构和 binlog 非常相似，只不过他多了一个 master.info 和 relay-log.info 的文件。

master.info 记录了上一次读取到 master 同步过来的 binlog 的位置，以及连接 master 和启动复制必须的所有信息。
relay-log.info 记录了文件复制的进度，下一个事件从什么位置开始，由 sql 线程负责更新。

数据库同步原理

 说完了原理，接下来说说常见的俩种主备模式以及实践吧。

主备模式与实践

常见的主备模式有俩种分别是 M-S 结构 和 双 M 结构，本篇文章介绍前者 – M-S 结构

M-S结构

M-S结构，两个节点，一个当主库、一个当备库，不允许两个节点互换角色。

M-S结构

在状态1中，客户端的读写都直接访问节点A，而节点B是A的备库，只是将A的更新都同步过来，到本地执行。这样可以保持节点B和A的数据是相同的。
当需要切换的时候，就切成状态2。这时候客户端读写访问的都是节点B，而节点A是B的备库。

实践

修改 master 配置

首先修改 master 中的 my.cnf 

1
vi /etc/my.cnf

在 my.cnf 中 [mysqld] 中添加

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
[mysqld]
bind-address=192.168.1.100
server-id=1
binlog-ignore-db = "mysql"
binlog-format = mixed 
log-bin=mysql-bin
datadir=/var/lib/mysql
innodb_flush_log_at_trx_commit=1
sync_binlog=1

#备注：
# server-id 服务器唯一标识。
# log_bin 启动MySQL二进制日志，即数据同步语句，从数据库会一条一条的执行这些语句。
# binlog_do_db 指定记录二进制日志的数据库，即需要复制的数据库名，如果复制多个数据库，重复设置这个选项即可。
# binlog_ignore_db 指定不记录二进制日志的数据库，即不需要复制的数据库名，如果有多个数据库，重复设置这个选项即可。
# 其中需要注意的是,binlog_do_db和binlog_ignore_db为互斥选项，一般只需要一个即可。

重启 master 的 mysql

保存之后，然后重启 MySQL

1
service restart mysql

创建同步帐号

接着以 root 帐号登录 master 的 mysql，创建一个用于同步的帐号，以下建立同步帐号名为 replication,密码为 password, slave 的 ip 是 192.168.1.101

1
2
3
4
5
mysql> CREATE USER replication@192.168.1.101;
mysql> GRANT REPLICATION SLAVE ON *.* TO replication@192.168.1.101 IDENTIFIED BY 'password';
mysql> flush privileges;
mysql> SHOW MASTER STATUS;
mysql> exit;

数据初始化

到此关于 master 的配置结束了，如果这个时候，要同步的数据库有数据，可以将数据 sql 文件，以便于前期的数据初始化

1
mysqldump –skip-lock-tables –all-databases –user=root –password –master-data > master.sql

修改 slave 的配置

接下来对 slave 的进行修改，首先修改 my.cnf，将下述内容加到 [mysqld] 区域:

1
2
3
4
5
6
[mysqld]
server-id=2
binlog-format=mixed
log_bin=mysql-bin
relay-log=mysql-relay-bin
log-slave-updates=1 # 默认是关闭的，这个时候 salve 从 master 复制的数据不会写入到 log-bin 日志文件中，开启后，则会写入

重启 slave 的 mysql

然后重启 mysql，重启之后，以 root 用户登录 MySQL, 新建要同步的数据库

1
2
3
4
service restart mysql
mysql -uroot -p
mysql> create database database-name;
mysql> exit;

导入 master 的数据

将 master 的导出的 sql 文件，加载到 slave 数据中，进行数据初始化

1
mysql -u root -p database-name < master.sql

设置跟踪主库日志文件

在开启 slave 服务之前，还需要进行一些设置

1
2
3
4
5
6
7
8
mysql> CHANGE MASTER TO 
    -> MASTER_HOST='192.168.1.100',
    -> MASTER_USER='replication',
    -> MASTER_PASSWORD='password', 
    -> MASTER_LOG_FILE='mysql-bin.000001', 
    -> MASTER_LOG_POS=500;
mysql> START SLAVE;
mysql> SHOW SLAVE STATUS \G;

每个参数的含义如下:

  • master_host 为主库IP地址

  • master_user 为主库用户名

  • master_password 为主库密码

  • master_log_file 为主库日志文件

  • master_log_pos 为主库日志所占位置

其中 MASTER_LOG_FILE 及 MASTER_LOG_POS 是在 Master 上在 MySQL 执行 “SHOW MASTER STATUS;” 的结果。

启动 slave 服务,并查看slave服务信息

1
2
mysql> START SLAVE;
mysql> SHOW SLAVE STATUS \G;

如果没有错误信息的话，就说明配置成功了。

参考资料

mysql 实现主备同步

MySQL中的binlog和relay-log结构完全详解

MySQL主从复制与双主互备

Mysql log 日志

mysql中binlog_format模式与配置详解

How To Set Up Master Slave Replication in MySQL

发表于 更新于

前言

scrapy 是 python 中一个优秀的爬虫框架，基于这个框架，用户可以快速构建自己的爬虫程序。框架涉及很多模块，其中有两个核心概念 items 和 item pipelines。

Items: 爬取的主要目标就是从非结构性的数据源提取结构性数据，例如网页。 Scrapy提供 Item 类来满足这样的需求。
Item 对象是种简单的容器，保存了爬取到得数据。 其提供了 类似于词典(dictionary-like) 的API以及用于声明可用字段的简单语法。

Item Pipeline:当Item在Spider中被收集之后，它将会被传递到Item Pipeline，一些组件会按照一定的顺序执行对Item的处理。
每个item pipeline组件(有时称之为“Item Pipeline”)是实现了简单方法的Python类。他们接收到Item并通过它执行一些行为，同时也决定此Item是否继续通过pipeline，或是被丢弃而不再进行处理。

``

Pipeline

通常情况下 scrapy 会有一个默认的 Pipeline：

1
2
3
class KnowledgeGraphPipeline:
    def process_item(self, item, spider):
        pass

正常对于传过来的数据，我们通过 item 对应的类型，可以进行判断然后执行逻辑

1
2
if isinstance(item, XXItem):
    # 代码逻辑

一旦当我们的 spider 多了，将所有的处理逻辑放到一个 pipeline 显然是不合理的，所以会希望每个 spider 都有一个专门的 pipeline，这个时候我们可以通过复写 spider 的默认配置就可以实现这样的目的

解决办法

1
2
3
4
5
6
7
class XXSpider(scrapy.Spider):
    name = 'xx'
    custom_settings = {
        'ITEM_PIPELINES': {
            'knowledge_graph.pipelines.xxxx': 200 # 这个值需要大于默认的 pipeline
        }
    }