-
Notifications
You must be signed in to change notification settings - Fork 54
/
__main__.py
64 lines (57 loc) · 2.18 KB
/
__main__.py
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
"""Simple aioftp-based server with one user (anonymous or not)"""
import asyncio
import logging
import socket
import argparse
import aioftp
parser = argparse.ArgumentParser(
prog="aioftp",
usage="%(prog)s [options]",
description="Simple aioftp-based server with one user (anonymous or not)."
)
parser.add_argument("--user", metavar="LOGIN", dest="login",
help="user name to login")
parser.add_argument("--pass", metavar="PASSWORD", dest="password",
help="password to login")
parser.add_argument("-d", metavar="DIRECTORY", dest="home",
help="the directory to share (default current directory)")
parser.add_argument("-q", "--quiet", action="store_true",
help="set logging level to 'ERROR' instead of 'INFO'")
parser.add_argument("--memory", action="store_true", help="use memory storage")
parser.add_argument("--host", default=None,
help="host for binding [default: %(default)s]")
parser.add_argument("--port", type=int, default=2121,
help="port for binding [default: %(default)s]")
parser.add_argument("--family", choices=("ipv4", "ipv6", "auto"),
default="auto",
help="Socket family [default: %(default)s]")
args = parser.parse_args()
print(f"aioftp v{aioftp.__version__}")
if not args.quiet:
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(name)s] %(message)s",
datefmt="[%H:%M:%S]:",
)
if args.memory:
user = aioftp.User(args.login, args.password, base_path="/")
path_io_factory = aioftp.MemoryPathIO
else:
if args.home:
user = aioftp.User(args.login, args.password, base_path=args.home)
else:
user = aioftp.User(args.login, args.password)
path_io_factory = aioftp.PathIO
family = {
"ipv4": socket.AF_INET,
"ipv6": socket.AF_INET6,
"auto": socket.AF_UNSPEC,
}[args.family]
server = aioftp.Server([user], path_io_factory=path_io_factory)
loop = asyncio.get_event_loop()
loop.run_until_complete(server.start(args.host, args.port, family=family))
try:
loop.run_forever()
except KeyboardInterrupt:
loop.run_until_complete(server.close())
loop.close()