File: kopano-spamhandler.py

package info (click to toggle)
kopanocore 8.7.0-3
  • links: PTS, VCS
  • area: main
  • in suites: bullseye, buster, sid
  • size: 15,400 kB
  • sloc: cpp: 175,422; python: 24,623; perl: 7,319; php: 6,056; sh: 2,172; makefile: 1,294; xml: 45; ansic: 1
file content (90 lines) | stat: -rwxr-xr-x 3,425 bytes parent folder | download
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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
#!/usr/bin/python3
# SPDX-License-Identifier: AGPL-3.0-only

# call (learning) "spam command" for items in junk folders, optionally deleting them

# usage: ./kopano-spamhandler.py (change kopano-spamhandler.cfg)

# XXX use python-kopano Config class

try:
    from ConfigParser import ConfigParser
except ImportError:
    from configparser import ConfigParser
import sys
import subprocess
import datetime
import kopano
delcounter = 0

def main():
    global delcounter
    learncounter = 0
    (users, allusers, remoteusers, autolearn, autodelete, deleteafter, spamcommand) = getconfig()
    z = kopano.Server()
    if allusers and not users:
        users = []
        for user in z.users(remote=remoteusers):
            users.append(user.name)
    for username in users:
        try:
            user = z.user(username)
            for item in user.store.junk.items():
                if autolearn:
                    if (not item.header('x-spam-flag')) or ( item.header('x-spam-flag') == 'NO'):
                        print("%s : untagged spam [Subject: %s]" % (user.name, item.subject))
                        try:
                            p = subprocess.Popen(spamcommand, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
                            learn, output_err = p.communicate(item.eml())
                        except:
                            print('failed to run [%s] [%s]' % (spamcommand, output_err))
                        if learn:
                            print("%s : learned [%s]" % (user.name, learn.rstrip('\n')))
                            delmsg = 'delete after learn'
                            deletejunk(user, item, delmsg)
                            learncounter += 1
                        continue
                if autodelete:
                    if item.received.date() < (datetime.date.today()-datetime.timedelta(days=deleteafter)):
                        delmsg = 'autodelete'
                        deletejunk(user, item, delmsg)
        except Exception as error:
            print("%s : Unable to open store/item : [%s] [%s]" % (username, username, error))
            continue
    print("Summary learned %d items, deleted %d items" % (learncounter, delcounter))


def deletejunk(user, item, delmsg):
    global delcounter
    try:
        user.store.junk.delete([item])
        print("%s : %s [Subject: %s]" % (user.name, delmsg, item.subject))
        delcounter += 1
    except Exception as error:
        print("%s : Unable to %s item [Subject: %s] [%s]" % (user.name, delmsg, item.subject, error))
        pass
    return


def getconfig():
    Config = ConfigParser()
    try:
        Config.read('kopano-spamhandler.cfg')
        users = Config.get('users', 'users')
        remoteusers = Config.getboolean('users', 'remoteusers')
        autolearn = Config.getboolean('learning', 'autolearn')
        autodelete = Config.getboolean('deleting', 'autodelete')
        deleteafter = Config.getint('deleting', 'deleteafter')
        spamcommand = Config.get('spamcommand', 'command')
        if not users:
            allusers = True
        else:
            allusers = False
            users = users.replace(" ", "").split(",")
        return (users, allusers, remoteusers, autolearn, autodelete, deleteafter, spamcommand)
    except:
        exit('Configuration error, please check kopano-spamhandler.cfg')


if __name__ == '__main__':
    main()