from multiprocessing import Process
count = 0
class MultiOneProcess(Process):
    def __init__(self,name):
        super().__init__()
        self.name = name
    def run(self) -> None:
        global count
        count += 1
        print('process name %s is running----count:%d'%(self.name, count))
if __name__ == '__main__':
    p_list = []
    for i in range(10):
        name = 'process_%d'%i
        p = MultiOneProcess(name = name)
        p.start()
        p_list.append(p)
    for p in p_list:
        p.join()
    print('this main process')

from multiprocessing import Process
count = 0
class MultiTwoProcess(Process):
    def __init__(self,name):
        super().__init__(target=self.do_fun)
        self.name = name
    def do_fun(self):
        global count
        count += 1
        print('process name %s is running----count:%d' % (name, count))
if __name__ == '__main__':
    p_list = []
    for i in range(10):
        name = 'process_%d'%i
        p = MultiTwoProcess(name)
        p.start()
        p_list.append(p)
    for p in p_list:
        p.join()
    print('this main process')

from multiprocessing import Process
count = 0
def do_fun(name):
    global count
    count += 1
    print('process name %s is running----count:%d' % (name, count))
 
if __name__ == '__main__':
    p_list = []
    for i in range(10):
        name = 'process_%d'%i
        p = Process(target=do_fun,args=(name,))
        p.start()
        p_list.append(p)
 
    for p in p_list:
        p.join()
    print('this main process')

from multiprocessing import Process, Queue
def send(q):
    while True:
        q.put('發(fā)送一個(gè)數(shù)據(jù)')
def receive1(q):
    while True:
        s = q.get()
        print('receive1:', s)
def receive2(q):
    while True:
        s = q.get()
        print('receive2:', s)
if __name__ == '__main__':
    q = Queue()
    p1 = Process(target=send,args=(q,))
    p2 = Process(target=receive1,args=(q,))
    p3 = Process(target=receive2,args=(q,))
    p1.start()
    p2.start()
    p3.start()

from multiprocessing import Process, Pipe
def fun2(p):
    while True:
        s = p.recv()
        print('接收一個(gè)數(shù)據(jù)：',s)
def fun1(p):
    while True:
        print('發(fā)送一個(gè)數(shù)據(jù)：pipe')
        p.send('pipe')
if __name__ == '__main__':
    pi1,pi2 = Pipe(duplex=True)
    p1 = Process(target=fun1,args=(pi1,))
    p2 = Process(target=fun2,args=(pi2,))
    p1.start()
    p2.start()

if __name__ == '__main__':
    #所有Excel的路徑
    all_paths = glob('../data/original_data/*')
    sysInfo_list = ['我通過了好友請(qǐng)求，現(xiàn)在你倆可以開始聊天了', '我通過了你的朋友驗(yàn)證請(qǐng)求，現(xiàn)在我們可以開始聊天了', '已通過你的朋友驗(yàn)證請(qǐng)求，現(xiàn)在可以開始聊天了', '不支持此消息，請(qǐng)?jiān)谑謾C(jī)上查看',
                    '微信紅包']
    interval = 25
    if len(all_paths)//interval * interval < len(all_paths):
        k = len(all_paths)//interval + 1
    else:
        k = len(all_paths) // interval
    #分段處理，每段25個(gè)Excel
    for i in range(k):
        paths = all_paths[i*interval:(i+1)*interval]
        if i*interval >= 100 and i*interval < 200:
            params = []
            for path in tqdm(paths):
                params.append((path, sysInfo_list))
            #多進(jìn)程處理——進(jìn)程池、以及進(jìn)度顯示
            with Pool(20) as p:
                res = list(tqdm(p.imap(extract_data, params), total=len(params), desc='extract_data'))
            all_df = []
            for dfs in res:
                if len(dfs) > 0:
                    all_df.extend(dfs)
            df = pd.concat(all_df, axis=0)
            save_path = '../data/weikong_clean_data_'+str(i*interval)+'_'+str(i*interval+len(paths)-1)+'.xlsx'
            writer = pd.ExcelWriter(save_path)
            df.to_excel(writer, index=False)
            writer.save()
            writer.close()

def save_mmi_train_data(queue):
    with open('../data/finetune_mmi_data/train.txt','w',encoding='utf-8') as f:
        while True:
            save_list = queue.get()
            if len(save_list) == 0:
                break
            for line in save_list:
                f.write(line)
def save_mmi_val_data(queue):
    with open('../data/finetune_mmi_data/val.txt','w',encoding='utf-8') as f:
        while True:
            save_list = queue.get()
            if len(save_list) == 0:
                break
            for line in save_list:
                f.write(line)
def get_funtine_data(paths):
    all_groups = []
    for path in tqdm(paths,desc='load data from excle'):
        df = pd.read_excel(path)
        df.dropna(inplace=True)
        df.drop_duplicates(inplace=True, keep='first')
        groups = list(df.groupby(by=['坐席id', '客戶微信id']))
        all_groups.extend(groups)
    print('len(all_groups)',len(all_groups))
    train, val = train_test_split(all_groups,test_size=10000/len(all_groups),random_state=1)
    print('len(train)', len(train))
    print('len(val)', len(val))
    train_std_path = '../data/finetune_std_data/train.txt'
    val_std_path = '../data/finetune_std_data/val.txt'
    train_mmi_queue = Queue()
    save_funtine_data(train, train_std_path,train_mmi_queue,save_mmi_train_data)
    val_mmi_queue = Queue()
    save_funtine_data(val, val_std_path, val_mmi_queue, save_mmi_val_data)
def save_funtine_data(groups,save_std_path,queue,fun):
    p = Process(target=fun,args=(queue,))
    p.start()
    with open(save_std_path,'w', encoding='utf-8') as f:
        for group in tqdm(groups, desc='find and save funtine dialogue datas'):
            new_df = group[1]
            df_roles = new_df['是否客服'].values.tolist()
            df_contents = new_df['消息內(nèi)容'].values.tolist()
            roles = []
            contents = []
            for role,content in zip(df_roles,df_contents):
                content = content.replace('\n', '')
                content = emoji.replace_emoji(content, '')
                if len(content) > 0 and content != "":
                    roles.append(role)
                    contents.append(content)
            save_list = []
            save_str = ""
            for index, role in enumerate(roles):
                content = contents[index].replace('\n','')
                content = emoji.replace_emoji(content, '')
                if content[-1] not in punctuations:
                    content += '；'
                if index == 0:
                    if role == "是":
                        save_str += "坐席："+content
                    else:
                        save_str += "客戶："+content
                else:
                    if role != roles[index-1]:
                        f.write(save_str[0:-1]+'\n')
                        save_list.append(save_str[0:-1]+'\n')
                        if role == "是":
                            save_str = "坐席：" + content
                        else:
                            save_str = "客戶：" + content
                    else:
                        save_str += content
            if len(save_str) > 1:
                save_list.append(save_str[0:-1] + '\n')
                f.write(save_str[0:-1]+'\n')
            f.write('\n')
            # 切片反轉(zhuǎn)
            save_list = save_list[::-1]
            save_list.append('\n')
            if len(save_list) > 0:
                queue.put(save_list)
    #注意傳入一個(gè)空值，讓倒序進(jìn)程結(jié)束
    queue.put([])
    p.join()

import a
import b
import c
from whole_queue import WholeQueue
import os
if __name__ == '__main__':
    print("main process:",os.getpid())
 
    whole_queue = WholeQueue()
    b_pool_size = 2
    c_pool_size = 6
    Module_list = [
        a.A(whole_queue,b_pool_size),
        b.B(whole_queue,b_pool_size,c_pool_size),
        c.C(whole_queue,c_pool_size)
    ]
 
    for p in Module_list:
        p.start()

class WholeQueue():
    def __init__(self):
        self.queues = dict()
    def register(self,queuename,queue):
        self.queues[queuename] =  queue

from multiprocessing import Process,Queue
import time
import random
import os
class A(Process):
    def __init__(self,whole_queue,b_pool_size):
        super().__init__(target=self.do_run)
        self.whole_queue = whole_queue
        self.b_pool_size = b_pool_size
        self.queue_list = []
        queue = Queue()
        self.whole_queue.register('A', queue)
        self.queue_list.append(queue)
        self.count  = 0
    def do_run(self):
        print("A.do_run process:", os.getpid())
        a_send_pro = Process(target = self.send)
        a_send_pro.start()
        a_receive_pro = Process(target = self.receive)
        a_receive_pro.start()
    def send(self):
        print("A.send process:", os.getpid())
        while True:
            time.sleep(0.001)
            self.whole_queue.queues['B_%d'%(self.count%self.b_pool_size)].put_nowait(self.count)
            self.count += 1
    def receive(self):
        print("A.receive process:", os.getpid())
        while True:
            rece = self.whole_queue.queues['A'].get()
            print(rece)
 

from multiprocessing import Process,Queue
import time
import random
import os
class B(Process):
    def __init__(self,whole_queue,b_pool_size,c_pool_size):
        super().__init__(target=self.do_run)
        self.whole_queue = whole_queue
        self.b_pool_size = b_pool_size
        self.c_pool_size = c_pool_size
        self.queue_list = []
        for i in range(self.b_pool_size):
            queue = Queue()
            self.whole_queue.register('B_%d'% i , queue)
            self.queue_list.append(queue)
        self.count  = 0
    def do_run(self):
        print("B.do_run process:", os.getpid())
        for i in range(self.b_pool_size):
            p = Process(target=self.component,args=(self.queue_list[i],))
            p.start()
    def component(self, queue):
        print("B.component process:", os.getpid())
        while True:
            time.sleep(0.01)
            info = queue.get()
            componext_info = 'component_' + str(info)
            self.whole_queue.queues['C_%d'%(info%self.c_pool_size)].put(componext_info)

from multiprocessing import Process,Queue
from model import Model
import time
import random
import os
class C(Process):
    def __init__(self,whole_queue,c_pool_size):
        super().__init__(target=self.do_run)
        self.whole_queue = whole_queue
        self.c_pool_size = c_pool_size
        self.queue_list = []
        for i in range(self.c_pool_size):
            queue = Queue()
            self.whole_queue.register('C_%d'% i , queue)
            self.queue_list.append(queue)
        # self.cache_queue = None
        # self.result_queue = None
        # self.infer_queue = None
    def do_run(self):
        cache_queue = Queue()
        result_queue = Queue()
        infer_queue = Queue()
        print("C.do_run process:", os.getpid())
        for i in range(self.c_pool_size):
            p = Process(target=self.receive,args=(self.queue_list[i], cache_queue,))
            p.start()
        cache_p = Process(target=self.cache,args=(cache_queue, infer_queue,))
        cache_p.start()
        predict_p = Process(target=self.predict,args=(infer_queue, result_queue))
        predict_p.start()
        while True:
            res = result_queue.get()
            for ele in res:
                self.whole_queue.queues['A'].put(ele)
    def receive(self, queue,cache_queue):
        print("C.receive process:", os.getpid())
        while True:
            info = queue.get()
            receive_info = 'receive_' + info
            cache_queue.put(receive_info)
    def cache(self,cache_queue, infer_queue):
        timeLast = time.time()
        print("C.cache process:", os.getpid())
        caches = []
        while True:
            data = cache_queue.get()
            caches.append(data)
            if len(caches) > 128 or time.time() - timeLast > 1:
                timeLast = time.time()
                infer_queue.put(caches)
                caches = []
    def predict(self,infer_queue, result_queue):
        print("C.predict process:", os.getpid())
        # 模型必須在這里初始化
        model = Model()
        while True:
            data = infer_queue.get()
            result = model(data)
            result = [ 'modelpredict_' + ele for ele in result]
            time.sleep(random.uniform(0.1,0.5))
            result_queue.put(result)