semvec/pypi-packages huggingface dataset:
kfk2es 1.1.0
get data from kafka, after processing send them to elasticsearch
Keywords: kafka elasticsearch logstash

# kafka to elasticsearch

从kafka读取数据，并处理后写入elastcsearch,只支持python3

## 使用方法

### 编写自己的处理程序


example.py

```python
from kfk2es import StreamProcess

def your_own_handler(event):
    """
    event的内容就是从kafka中获取的每条数据
    """
    print(event)
    #return ('esindex-%Y%m%d', '_log', event)
    #return (None, '_log', event)
    return event # 返回的结果会发到es中，如果没有返回，则不发

if __name__ == '__main__':
    pipe = StreamProcess(configfile='config.yml')
    # from myconf import Conf
    # conf = Conf('config.yml')
    # pipe = StreamProcess(configfile=conf)
    pipe.handler = your_own_handler
    pipe.run()

```

StreamProcess的configfile参数可以是一个配置文件路径，也可以是包含必要参数的字典。

如果不添加任何处理程序，那么从kafka读取的数据会直接尝试往elasticsearch发送

自己添加处理程序，只需要编写函数，并按照如下格式进行返回:

- 返回一个字典， 那么字典中的数据会被传到es中
- 返回一个3个元素的tuple， 那么第一个元素是es的索引，第二个元素则是es的类型，第三个元素为需要导入es的数据。 
通过数据直接返回的索引和类型要优先于配置文件中的配置。如果只希望动态设置其中一个，另一个使用默认，那么不需要定义的设置为None即可
- 如果不需要传输到es中，只要不返回任何数据就可以了。这样可以只测试kafka的连通性

### 编辑配置文件

配置文件可以是json或者yaml格式，以json,yaml或者yml结尾

可以使用多个kafka作为输入，elasticsearch只有一个输出

#### yaml格式例子

config.yml

```yaml
kafka:
  - topic: top
    bootstrap_servers:
      - server1:9092
      - server2:9092
      - server3:9092
    group_id: group_id
    sasl_plain_username: user
    sasl_plain_password: password

elasticsearch:
  params:               #parms for elasticsearch
    hosts:
      - http://user:pass@10.1.0.1:9200
  index: test-%Y%m%d
  type: log             # optional(default: "log")
  cache_size: 150       # optional(default: 150)
  timeout: 1            # optional(default: 1)

```

需要注意的是elasticsearch选项中的parms选项包含了elasticsearch库中Elasticsearch的参数. 而其他参数是StreamProcess需要的
xfinity-gateway 0.0.4
Query an Xfinity Gateway

# Query an Xfinity Gateway

This library allows an Xfinity Gateway to be queried for connected devices. It uses the `requests` library to scrape the web interface of the gateway.

## Usage
```python
from xfinity_gateway import XfinityGateway

# Connect to gateway
gateway = XfinityGateway('10.0.0.1')

# Get list of all connected devices' MAC addresses
gateway.scan_devices()

# Get specific device's name
gateway.get_device_name('mac address of device here')
```

## Thoughts

I have only tested on my own Xfinity-provided gateway:

| | |
| --- | --- |
| Model: | TG1682G |
| Hardware Revision: | 9.0 |

If your gateway's web interface looks like the following, this library will likely work:
![xfinity interface](https://i.imgur.com/4Jlcshl.png)

Please open a Github issue if you encounter any problems. Thanks!
aist 0.2.3
AIST = AI Service Tools

AIST = AI Service Tools

------------

公众号：人工智能AIST

这个python模块需要配合我们的微信公众号使用。目前正在测试中。请联系 me@feimax.com 开通功能。


目前支持的功能有：

团队协作的图片收集
-----
首先需要一个人开通这项功能，然后会自动产生一个加入任务的二维码，分享这个二维码，就可以让其他人加入到这个图片采集任务。收集人只需要将图片发送到公众号，就会自动上传到云端存储。

任务发起者可以获取下载密钥，在aist模块中，通过设置下载密钥，然后设置下载路径，就可以将云端的图片下载到本地了。


OCR扫描
-----
使用者只需要将图片发送给本公众号，就可以得到OCR的结果了。


数据通道
-----
该功能用于将科研代码中的数据实时推送到微信端，通过我们的公众号实时推送到你的微信。


安装方法
-------
```
# 升级 pip 到最新版本
python -m pip install --upgrade pip
# 从 pip 安装 aist 模块
pip install aist
```


使用方法
----------
获取图片
------------
```python
from aist.pic import Download

dn = Download('PC80000000')

# 获取该图片收集任务的图片列表，包含图片在云端的名称，数据大小和贡献者的wx_openid
print(dn.ls())
# 获取该图片收集任务的图片列表，包含
dn.ls('1.csv')

# 下载所有的图片文件到 test 文件夹下去
dn.all('test')
```

数据通道
-----------
```python
from aist.msg Import Msg

# 这里要填写从 人工智能AIST 公众号里获取的数据通道的密钥。
# 您需要练习 人工智能AIST 公众号来获取该项功能。
msg = Msg('MS0000000')

# push 方法直接将您想得到的信息直接通过我们的微信公众号推送给您。但是这种方法有两个限制：
# 1、每天的推送不能超过1500条，请仅仅推送重要的信息。
# 2、微信只允许在48小时之内发过信息给公众号的用户接受到公众号的推送。
#    也就是说，如果您超过48小时没有给我们的公众号发送过任何信息，我们将没有权限推送信息给您。
#    所以，为了确保您的使用，请经常发信息给公众号。
msg.push('这是一条推送信息')

# put 方法是将数据或者信息暂存，然后您任何时间都可以到我们的 人工智能AIST 公众号里去查询。
# 这种方法暂时没有条数限制，如果不是重要的信息，最好使用该方法。该方法同时会帮你记录数据暂存的时间，使用GMT标准时间。请自行换算时区
msg.put('这是一条正常信息')
```