semvec/pypi-packages huggingface dataset:
ccprocesspool 2.1
Process pool to add/check/stop all processes in pool
License: MIT

# Install

```
pip3 install ccprocesspool --upgrade --user
```


# Process Pool Usage

```
#!/usr/bin/env python3
# _*_ coding:utf-8 _*_
import time
from ccprocesspool import CCProcessPool


class CCProcessPoolTestJob(object):

    @staticmethod
    def count_instant_to_10():
        for i in range(0, 10):
            print('instant', i)

    @staticmethod
    def count_quick_to(param: int):
        for i in range(0, param):
            print('quick', i)

    @staticmethod
    def count_slow_to(param: int):
        for i in range(0, param):
            print('slow', i)
            time.sleep(1)


if __name__ == '__main__':
    p_m = CCProcessPool('test_pool')
    p_m.add_job(CCProcessPoolTestJob.count_instant_to_10, None)
    p_m.add_job(CCProcessPoolTestJob.count_quick_to, (10, ))
    p_m.add_job(CCProcessPoolTestJob.count_slow_to, (10, ))
    p_m.add_job(CCProcessPoolTestJob.count_slow_to, (10, ))
    p_m.check_job()
    while True:
        time.sleep(2)
        p_m.check_job()
        pass
```

# Restartable Process Usage

```
p_m = CCRestartableProcess('test_process', CCRestartableProcessTestJob.count_slow_to_10)
p_m.start()
while True:
    time.sleep(5)
    p_m.restart()
    pass
```
ccdingtalk 1.5
Send dingtalk webhook with simple interface and load balance
License: MIT

# Install

```
pip3 install ccdingtalk --upgrade --user
```


# Send Message using CCDingTalkWebhookMessage

```
from ccdingtalk import CCDingTalkWebhookMessage

chat_bot = CCDingTalkWebhookMessage(webhook_url)
r = chat_bot.send_text(msg=msg, at_mobiles=at_mobiles)
```


# Send with several webhooks (auto load balance) using CCDingTalkWebhookMessageQueue


```
from ccdingtalk import CCDingTalkWebhookMessageQueue

def __webhook_urls() -> list:
    api = 'https://oapi.dingtalk.com/robot/send?access_token='
    tokens = [
        'token1',
        'token2',
        'token3',
        'token4',
        'token5',
    ]
    results = []
    for token in tokens:
        results.append(api + token)
    return results


__shared_msg_queue = CCDingTalkWebhookMessageQueue(__webhook_urls())


def send_ding_msg(msg: str, at_mobile=''):
    __shared_msg_queue.add_message(msg, [at_mobile])
```

# Reply Message using CCDingTalkReply


```
from ccdingtalk import CCDingTalkReply

class DingBotRequestHandler(BaseHTTPRequestHandler):

    def do_GET(self):
		response = CCDingTalkReply.text_msg("This is GET request")
		response_bytes = response.encode()
		self.send_header("Content-Type", "text/html; charset=utf-8")
		self.send_header("Content-Length", str(len(response_bytes)))
		self.end_headers()
		self.wfile.write(response_bytes)

    def do_POST(self):
		response = CCDingTalkReply.text_msg("This is POST request")
		response_bytes = response.encode()
		self.send_header("Content-Type", "text/html; charset=utf-8")
		self.send_header("Content-Length", str(len(response_bytes)))
		self.end_headers()
		self.wfile.write(response_bytes)
```